104 lines
4.6 KiB
Python
104 lines
4.6 KiB
Python
import argparse
|
|
from pathlib import Path
|
|
|
|
from .core.config import Settings
|
|
from .core.utils import save_to_json
|
|
from .encar import EncarScraper, EncarFilters, CAR_TYPE_ALL, CAR_TYPE_DOMESTIC, CAR_TYPE_IMPORT
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Encar scraper CLI")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
default_output_dir = Path("artifacts/json")
|
|
|
|
subparsers.add_parser("init-db", help="Create DB tables")
|
|
|
|
# collect-listing
|
|
collect_parser = subparsers.add_parser("collect-listing", help="Collect Encar listing")
|
|
collect_parser.add_argument("--limit", type=int, default=None, help="Max items to collect")
|
|
collect_parser.add_argument("--car-type", choices=["all", "domestic", "import"], default="all",
|
|
help="all=все, domestic=корейские, import=импорт")
|
|
collect_parser.add_argument("--manufacturer", default=None, help="Filter by manufacturer (e.g. BMW, 현대)")
|
|
collect_parser.add_argument("--year-from", type=int, default=None, help="Min year")
|
|
collect_parser.add_argument("--year-to", type=int, default=None, help="Max year")
|
|
collect_parser.add_argument("--price-max", type=int, default=None, help="Max price in 만원 (10k KRW)")
|
|
collect_parser.add_argument("--output", default=str(default_output_dir / "encar_listing.json"))
|
|
|
|
# scrape-vehicle
|
|
scrape_parser = subparsers.add_parser("scrape-vehicle", help="Fetch single vehicle data")
|
|
scrape_parser.add_argument("vehicle_url")
|
|
scrape_parser.add_argument("--output", default=str(default_output_dir / "encar_vehicle_detail.json"))
|
|
|
|
# sync-vehicle
|
|
sync_parser = subparsers.add_parser("sync-vehicle", help="Sync single vehicle into DB")
|
|
sync_parser.add_argument("vehicle_url")
|
|
sync_parser.add_argument("--lane", default="encar")
|
|
sync_parser.add_argument("--output", default=str(default_output_dir / "encar_sync_vehicle.json"))
|
|
|
|
# sync-listing
|
|
sync_listing_parser = subparsers.add_parser("sync-listing", help="Collect listing and sync all to DB")
|
|
sync_listing_parser.add_argument("--limit", type=int, default=None, help="Max items to sync")
|
|
sync_listing_parser.add_argument("--car-type", choices=["all", "domestic", "import"], default="all")
|
|
sync_listing_parser.add_argument("--manufacturer", default=None, help="Filter by manufacturer")
|
|
sync_listing_parser.add_argument("--year-from", type=int, default=None)
|
|
sync_listing_parser.add_argument("--year-to", type=int, default=None)
|
|
sync_listing_parser.add_argument("--price-max", type=int, default=None)
|
|
sync_listing_parser.add_argument("--lane", default="encar")
|
|
sync_listing_parser.add_argument("--only-new", choices=["true", "false"], default="true")
|
|
sync_listing_parser.add_argument("--output", default=str(default_output_dir / "encar_sync_listing.json"))
|
|
|
|
return parser
|
|
|
|
|
|
def _build_filters_from_args(args) -> EncarFilters:
|
|
"""Строит фильтры из аргументов CLI."""
|
|
car_type_map = {"all": CAR_TYPE_ALL, "domestic": CAR_TYPE_DOMESTIC, "import": CAR_TYPE_IMPORT}
|
|
return EncarFilters(
|
|
car_type=car_type_map.get(getattr(args, "car_type", "all"), CAR_TYPE_ALL),
|
|
manufacturer=getattr(args, "manufacturer", None),
|
|
year_from=getattr(args, "year_from", None),
|
|
year_to=getattr(args, "year_to", None),
|
|
price_to=getattr(args, "price_max", None),
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
import sys
|
|
import logging
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
|
stream=sys.stderr,
|
|
)
|
|
|
|
parser = build_parser()
|
|
args = parser.parse_args()
|
|
scraper = EncarScraper()
|
|
|
|
if args.command == "init-db":
|
|
result = scraper.init_db()
|
|
print(f"DB initialized: {result}")
|
|
return
|
|
elif args.command == "collect-listing":
|
|
filters = _build_filters_from_args(args)
|
|
result = scraper.collect_listing(limit=args.limit, filters=filters)
|
|
elif args.command == "scrape-vehicle":
|
|
result = scraper.scrape_vehicle_detail(args.vehicle_url)
|
|
elif args.command == "sync-vehicle":
|
|
result = scraper.sync_vehicle(args.vehicle_url, lane=args.lane)
|
|
elif args.command == "sync-listing":
|
|
filters = _build_filters_from_args(args)
|
|
only_new = args.only_new == "true"
|
|
result = scraper.sync_listing(limit=args.limit, filters=filters, lane=args.lane, only_new=only_new)
|
|
else:
|
|
raise SystemExit(f"Unknown command: {args.command}")
|
|
|
|
save_to_json(result, Path(args.output))
|
|
print(f"Saved to {Path(args.output).resolve()}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|