from django.core.management.base import BaseCommand from typing import Optional, List import json from manual_agregator.server_monitor.engine import ServerMonitorEngine class Command(BaseCommand): help = "Server monitor CLI. All operational logic lives in manual_agregator.server_monitor.engine.\n\n" help += "Actions:\n" help += " health Snapshot portal health metrics (default).\n" help += " run Run all checks for a single portal (network, parser, freshness).\n" help += " debug-dump Dump HTML/selectors/error bundles for debugging.\n" help += " lint Lint selector configs against model fields.\n" def add_arguments(self, parser): # Subcommands subparsers = parser.add_subparsers(dest="action", metavar="action", help="Action to perform") # health p_health = subparsers.add_parser("health", help="Snapshot portal health metrics") p_health.add_argument("--name", "-n", dest="names", type=str, help="Comma-separated portal names to include (omit with --all)") p_health.add_argument("--all", dest="all_portals", action="store_true", help="Include all portals") p_health.add_argument("--limit", dest="limit", type=int, default=25, help="Max portals in output") p_health.add_argument("--threshold-error", dest="thr_error", type=float, default=0.3, help="Error-rate threshold for unhealthy (0..1)") p_health.add_argument("--json", dest="json_out", type=str, help="Optional file to write JSON snapshot") # run (single portal) p_run = subparsers.add_parser("run", help="Run all checks for a single portal") p_run.add_argument("--portal", "-p", required=True, help="Portal name (exact)") p_run.add_argument("--hours", type=int, default=24, help="Lookback window for checks (where applicable)") p_run.add_argument("--json", dest="json_out", type=str, help="Optional file to write JSON output") # debug-dump p_dump = subparsers.add_parser("debug-dump", help="Write debug bundles (HTML, selectors, errors)") p_dump.add_argument("--name", "-n", dest="names", type=str, help="Comma-separated portal names (omit with --all)") p_dump.add_argument("--all", dest="all_portals", action="store_true", help="Process all portals") p_dump.add_argument("--limit", type=int, default=25, help="Max pages per portal") p_dump.add_argument("--only-unparsed", action="store_true", help="Only pages without parsed ads") p_dump.add_argument("--only-errors", action="store_true", help="Only pages with errors") p_dump.add_argument("--out", dest="out_dir", type=str, default="debug/manual_dump", help="Output directory root") p_dump.add_argument("--check-selectors", action="store_true", help="Count selector hits with BeautifulSoup") p_dump.add_argument("--json", dest="json_out", type=str, help="Optional file to write summary JSON") # lint p_lint = subparsers.add_parser("lint", help="Lint selector configs vs model fields") p_lint.add_argument("--name", "-n", dest="names", type=str, help="Comma-separated portal names (omit with --all)") p_lint.add_argument("--all", dest="all_portals", action="store_true", help="Lint all portals") p_lint.add_argument("--json", dest="json_out", type=str, help="Optional file to write results JSON") # Default behavior: if no subcommand given, treat as 'health' parser.set_defaults(action="health") def handle(self, *args, **opts): action = opts.get("action") or "health" engine = ServerMonitorEngine() # Common portal selection helper def parse_portals(names_arg: Optional[str], all_flag: bool) -> Optional[List[str]]: return None if all_flag else ([n.strip() for n in (names_arg or '').split(',') if n.strip()] or None) if action == "health": names_arg: Optional[str] = opts.get("names") all_portals: bool = bool(opts.get("all_portals")) json_out: Optional[str] = opts.get("json_out") thr_error: float = float(opts.get("thr_error") or 0.3) limit: int = int(opts.get("limit") or 25) portals = parse_portals(names_arg, all_portals) data = engine.health_snapshot(portals=portals, limit=limit, thr_error=thr_error) # Console summary or JSON output if json_out: with open(json_out, "w", encoding="utf-8") as fh: json.dump(data, fh, ensure_ascii=False, indent=2) self.stdout.write(self.style.SUCCESS(f"Saved JSON to {json_out}")) else: for row in data.get("portals", []): self.stdout.write( f"- {row['portal']}: pages={row['total_pages']} parsed={row['parsed_pages']} errors={row['error_pages']} " f"parse_rate={row['parse_rate']:.2%} error_rate={row['error_rate']:.2%} healthy={row['healthy']} last_ad={row['last_ad_created']}" ) return if action == "run": portal: str = opts.get("portal") hours: int = int(opts.get("hours") or 24) json_out: Optional[str] = opts.get("json_out") data = engine.run_all(portal=portal, hours=hours) if json_out: with open(json_out, "w", encoding="utf-8") as fh: json.dump(data, fh, ensure_ascii=False, indent=2) self.stdout.write(self.style.SUCCESS(f"Saved JSON to {json_out}")) else: self.stdout.write(json.dumps(data, ensure_ascii=False, indent=2)) return if action == "debug-dump": names_arg: Optional[str] = opts.get("names") all_portals: bool = bool(opts.get("all_portals")) limit: int = int(opts.get("limit") or 25) only_unparsed: bool = bool(opts.get("only_unparsed")) only_errors: bool = bool(opts.get("only_errors")) out_dir: str = opts.get("out_dir") or "debug/manual_dump" check_selectors: bool = bool(opts.get("check_selectors")) json_out: Optional[str] = opts.get("json_out") portals = parse_portals(names_arg, all_portals) summary = engine.debug_dump( portals=portals, all_portals=all_portals, limit=limit, only_unparsed=only_unparsed, only_errors=only_errors, out_dir=out_dir, check_selectors=check_selectors, ) if json_out: with open(json_out, "w", encoding="utf-8") as fh: json.dump(summary, fh, ensure_ascii=False, indent=2) self.stdout.write(self.style.SUCCESS(f"Saved JSON to {json_out}")) else: self.stdout.write(self.style.SUCCESS(f"Bundles: {summary.get('bundles', 0)} -> {summary.get('out')}")) return if action == "lint": names_arg: Optional[str] = opts.get("names") all_portals: bool = bool(opts.get("all_portals")) json_out: Optional[str] = opts.get("json_out") portals = parse_portals(names_arg, all_portals) results = engine.lint_selectors(portals=portals, all_portals=all_portals) if json_out: with open(json_out, "w", encoding="utf-8") as fh: json.dump(results, fh, ensure_ascii=False, indent=2) self.stdout.write(self.style.SUCCESS(f"Saved JSON to {json_out}")) else: for res in results: portal = res.get("portal") unknown = len(res.get("unknown_field_paths", [])) bad = len(res.get("bad_property_paths", [])) if unknown or bad: self.stdout.write(self.style.WARNING(f"{portal}: {unknown} unknown fields, {bad} bad props")) else: self.stdout.write(self.style.SUCCESS(f"{portal}: OK")) return # Default fallback self.stdout.write(self.style.ERROR("Unknown action. Use one of: health, run, debug-dump, lint"))