146 lines
5.5 KiB
Python
146 lines
5.5 KiB
Python
"""
|
|
Long-running scheduler — watches for new JSON files and syncs to MySQL.
|
|
|
|
Modes:
|
|
--watch : File watcher, syncs immediately when new JSON appears (default)
|
|
--cron : One-shot sync, meant to be called by system cron/launchd
|
|
--daemon : Combined: runs initial sync + watches for changes
|
|
"""
|
|
|
|
import time
|
|
import sys
|
|
import signal
|
|
import argparse
|
|
import logging
|
|
import importlib
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from typing import Any, cast
|
|
|
|
try:
|
|
from . import aps_db_sync as aps_db_sync_module
|
|
except ImportError:
|
|
aps_db_sync_module = importlib.import_module("aps_db_sync")
|
|
|
|
APSSyncer = aps_db_sync_module.APSSyncer
|
|
db_config_default = cast(dict[str, str | int], aps_db_sync_module.DB_CONFIG)
|
|
json_dir = cast(Path, aps_db_sync_module.JSON_DIR)
|
|
default_sync_target = cast(str, aps_db_sync_module.SYNC_TARGET_ALL)
|
|
valid_sync_targets = cast(set[str], aps_db_sync_module.VALID_SYNC_TARGETS)
|
|
|
|
LOG_FORMAT = "%(asctime)s [%(levelname)s] %(message)s"
|
|
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
|
|
logger = logging.getLogger("aps_scheduler")
|
|
|
|
DEFAULT_WATCH_INTERVAL_SECONDS = 30
|
|
watch_interval_seconds = DEFAULT_WATCH_INTERVAL_SECONDS
|
|
PROCESSED_MARKER_DIR = json_dir / ".aps_sync_processed"
|
|
|
|
|
|
def _update_watch_interval(value: int):
|
|
global watch_interval_seconds
|
|
watch_interval_seconds = value
|
|
|
|
|
|
class SyncScheduler:
|
|
def __init__(self, db_config: dict[str, str | int] | None = None, sync_target: str = default_sync_target):
|
|
self.db_config: dict[str, str | int] = db_config or db_config_default
|
|
self.sync_target: str = sync_target
|
|
self.running: bool = True
|
|
PROCESSED_MARKER_DIR.mkdir(exist_ok=True)
|
|
_ = signal.signal(signal.SIGINT, self._shutdown)
|
|
_ = signal.signal(signal.SIGTERM, self._shutdown)
|
|
|
|
def _shutdown(self, signum: int, frame: object | None):
|
|
logger.info("Shutdown signal received, stopping...")
|
|
self.running = False
|
|
|
|
def _marker_path(self, json_path: Path) -> Path:
|
|
return PROCESSED_MARKER_DIR / f"{json_path.stem}.synced"
|
|
|
|
def _is_processed(self, json_path: Path) -> bool:
|
|
marker = self._marker_path(json_path)
|
|
if not marker.exists():
|
|
return False
|
|
marker_mtime = marker.stat().st_mtime
|
|
json_mtime = json_path.stat().st_mtime
|
|
return marker_mtime >= json_mtime
|
|
|
|
def _mark_processed(self, json_path: Path):
|
|
marker = self._marker_path(json_path)
|
|
_ = marker.write_text(datetime.now().isoformat())
|
|
|
|
def find_unprocessed_files(self) -> list[Path]:
|
|
pattern = "aps_aliyun_customers_with_bills_*.json"
|
|
all_files = sorted(json_dir.glob(pattern), key=lambda p: p.stat().st_mtime)
|
|
return [f for f in all_files if not self._is_processed(f)]
|
|
|
|
def sync_file(self, json_path: Path) -> bool:
|
|
logger.info("Syncing: %s", json_path.name)
|
|
try:
|
|
syncer = APSSyncer(db_config=cast(Any, self.db_config))
|
|
_ = syncer.sync_from_json(str(json_path), sync_target=self.sync_target)
|
|
self._mark_processed(json_path)
|
|
return True
|
|
except Exception as e:
|
|
logger.error("Sync failed for %s: %s", json_path.name, e)
|
|
return False
|
|
|
|
def run_once(self):
|
|
unprocessed = self.find_unprocessed_files()
|
|
if not unprocessed:
|
|
logger.info("No unprocessed JSON files found")
|
|
return 0
|
|
count = 0
|
|
for f in unprocessed:
|
|
if self.sync_file(f):
|
|
count += 1
|
|
logger.info("Processed %d/%d files", count, len(unprocessed))
|
|
return count
|
|
|
|
def run_watch(self):
|
|
logger.info("Watching %s for new JSON files (interval=%ds)", json_dir, watch_interval_seconds)
|
|
_ = self.run_once()
|
|
while self.running:
|
|
time.sleep(watch_interval_seconds)
|
|
unprocessed = self.find_unprocessed_files()
|
|
for f in unprocessed:
|
|
if not self.running:
|
|
break
|
|
_ = self.sync_file(f)
|
|
logger.info("Watcher stopped")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="APS Sync Scheduler")
|
|
_ = parser.add_argument("--mode", choices=["watch", "cron", "daemon"], default="watch",
|
|
help="watch=file watcher, cron=one-shot, daemon=watch with initial sync")
|
|
_ = parser.add_argument("--host", default=db_config_default["host"])
|
|
_ = parser.add_argument("--port", type=int, default=db_config_default["port"])
|
|
_ = parser.add_argument("--user", default=db_config_default["user"])
|
|
_ = parser.add_argument("--password", default=db_config_default["password"])
|
|
_ = parser.add_argument("--database", default=db_config_default["database"])
|
|
_ = parser.add_argument("--interval", type=int, default=watch_interval_seconds,
|
|
help="Watch interval in seconds")
|
|
_ = parser.add_argument("--sync-target", choices=sorted(valid_sync_targets), default=default_sync_target,
|
|
help="选择同步对象: all/customer/order/orderdetails/bills")
|
|
args = parser.parse_args()
|
|
|
|
_update_watch_interval(args.interval)
|
|
|
|
config: dict[str, str | int] = {
|
|
"host": args.host, "port": args.port, "user": args.user,
|
|
"password": args.password, "database": args.database, "charset": "utf8mb4",
|
|
}
|
|
scheduler = SyncScheduler(db_config=config, sync_target=args.sync_target)
|
|
|
|
if args.mode == "cron":
|
|
count = scheduler.run_once()
|
|
sys.exit(0 if count >= 0 else 1)
|
|
else:
|
|
scheduler.run_watch()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|