1f0dcb1ddb
- Frontend: Next.js 14 with TypeScript - Backend: FastAPI with SQLAlchemy - Agent: Carmodoo sync agent - Deployment: Docker Compose based staging/production setup - Scripts: Automated deployment with rollback support 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
98 lines
3.1 KiB
Python
98 lines
3.1 KiB
Python
"""성능점검표 직접 가져오기 (checkNum 직접 입력)"""
|
|
import asyncio
|
|
import sys
|
|
import os
|
|
import httpx
|
|
from lxml import html as lxml_html
|
|
|
|
os.chdir(os.path.dirname(os.path.abspath(__file__)))
|
|
sys.path.insert(0, '.')
|
|
|
|
from app.database import SessionLocal
|
|
from app.models import Car, CarPerformanceCheck
|
|
|
|
async def fetch_performance_check_direct(check_num: str):
|
|
"""checkNum으로 직접 성능점검표 가져오기"""
|
|
print(f"Fetching performance check: {check_num}")
|
|
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
perf_url = "https://ck.carmodoo.com/carCheck/carmodooPrint.do"
|
|
params = {"print": "0", "checkNum": check_num}
|
|
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 6.2; WOW64) AppleWebKit/537.36',
|
|
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
|
|
'Accept-Language': 'ko-KR,ko;q=0.9',
|
|
}
|
|
|
|
response = await client.get(perf_url, params=params, headers=headers)
|
|
|
|
if response.status_code == 200:
|
|
html = response.content.decode('utf-8')
|
|
print(f"Got HTML response: {len(html)} bytes")
|
|
|
|
# 간단한 파싱
|
|
tree = lxml_html.fromstring(html)
|
|
|
|
# 차명 추출
|
|
car_name = tree.xpath('//th[contains(text(), "차명")]/following-sibling::td/text()')
|
|
if car_name:
|
|
print(f"Car name: {car_name[0].strip()}")
|
|
|
|
# 등록번호 추출
|
|
car_number = tree.xpath('//th[contains(text(), "자동차등록번호")]/following-sibling::td/text()')
|
|
if car_number:
|
|
print(f"Car number: {car_number[0].strip()}")
|
|
|
|
# 주행거리 추출
|
|
mileage_text = tree.xpath('//*[contains(text(), "현재 주행거리")]/text()')
|
|
print(f"Mileage raw: {mileage_text}")
|
|
|
|
return html
|
|
else:
|
|
print(f"Error: {response.status_code}")
|
|
return None
|
|
|
|
async def save_to_db(car_id: int, check_num: str, html: str):
|
|
"""성능점검표 DB에 저장"""
|
|
db = SessionLocal()
|
|
|
|
try:
|
|
# 기존 데이터 삭제
|
|
existing = db.query(CarPerformanceCheck).filter(CarPerformanceCheck.car_id == car_id).first()
|
|
if existing:
|
|
db.delete(existing)
|
|
db.commit()
|
|
print(f"Deleted existing performance check for car {car_id}")
|
|
|
|
# 새 데이터 저장
|
|
perf_check = CarPerformanceCheck(
|
|
car_id=car_id,
|
|
check_number=check_num,
|
|
raw_html=html,
|
|
)
|
|
db.add(perf_check)
|
|
db.commit()
|
|
print(f"Saved performance check {check_num} for car {car_id}")
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
async def main():
|
|
# 성능점검번호 (사용자가 제공한 K5 차량)
|
|
check_num = "7400044430"
|
|
|
|
# 성능점검표 가져오기
|
|
html = await fetch_performance_check_direct(check_num)
|
|
|
|
if html:
|
|
# 가장 최근 차량(K5)에 저장 - ID 6
|
|
car_id = 6
|
|
await save_to_db(car_id, check_num, html)
|
|
print("Done!")
|
|
else:
|
|
print("Failed to fetch performance check")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|