118 lines
3.5 KiB
Python
118 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
# Direct execution from /app/scripts must resolve the repository's backend
|
|
# package before similarly named installed packages.
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
BACKEND_ROOT = ROOT / "backend" if (ROOT / "backend" / "app").is_dir() else ROOT
|
|
if str(BACKEND_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(BACKEND_ROOT))
|
|
|
|
from app.db.session import SessionLocal
|
|
from app.models import Project
|
|
|
|
|
|
CANONICAL_PROJECT_NAMES = frozenset(
|
|
{
|
|
"Kempen Regional Workbench",
|
|
"Mol Municipality Workbench",
|
|
}
|
|
)
|
|
|
|
TECHNICAL_PROJECT_PATTERNS = (
|
|
re.compile(r"^GeoIntel Detection Quality Matrix", re.IGNORECASE),
|
|
re.compile(r"^GeoIntel hard-negative", re.IGNORECASE),
|
|
re.compile(r"^GeoIntel Detection Calibration", re.IGNORECASE),
|
|
re.compile(r"^GeoIntel Real Data Validation", re.IGNORECASE),
|
|
re.compile(r"^GeoIntel Operational YOLO .* Smoke", re.IGNORECASE),
|
|
re.compile(r"^GeoIntel Demo - Building QA$", re.IGNORECASE),
|
|
re.compile(r"^GeoIntel training", re.IGNORECASE),
|
|
re.compile(r"^Mol Building QA", re.IGNORECASE),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ArchivePlan:
|
|
project_ids: tuple[UUID, ...]
|
|
names: tuple[str, ...]
|
|
|
|
@property
|
|
def count(self) -> int:
|
|
return len(self.project_ids)
|
|
|
|
|
|
def is_technical_project_name(name: str) -> bool:
|
|
normalized = name.strip()
|
|
if normalized in CANONICAL_PROJECT_NAMES:
|
|
return False
|
|
return any(pattern.search(normalized) for pattern in TECHNICAL_PROJECT_PATTERNS)
|
|
|
|
|
|
def build_archive_plan(db: Session) -> ArchivePlan:
|
|
rows = (
|
|
db.query(Project)
|
|
.filter(Project.status == "active")
|
|
.order_by(Project.created_at.asc())
|
|
.all()
|
|
)
|
|
selected = [row for row in rows if is_technical_project_name(row.name)]
|
|
return ArchivePlan(
|
|
project_ids=tuple(row.id for row in selected),
|
|
names=tuple(row.name for row in selected),
|
|
)
|
|
|
|
|
|
def apply_archive_plan(db: Session, plan: ArchivePlan) -> int:
|
|
if not plan.project_ids:
|
|
return 0
|
|
updated = (
|
|
db.query(Project)
|
|
.filter(Project.id.in_(plan.project_ids), Project.status == "active")
|
|
.update({Project.status: "archived"}, synchronize_session=False)
|
|
)
|
|
db.commit()
|
|
return int(updated)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description=(
|
|
"Archive allowlisted GeoIntel operator and benchmark projects. "
|
|
"Dry-run is the default; pass --apply to persist status changes."
|
|
)
|
|
)
|
|
parser.add_argument("--apply", action="store_true", help="Persist status='archived'.")
|
|
parser.add_argument(
|
|
"--show-names",
|
|
action="store_true",
|
|
help="Include every matched project name in the JSON output.",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
with SessionLocal() as db:
|
|
plan = build_archive_plan(db)
|
|
archived_count = apply_archive_plan(db, plan) if args.apply else 0
|
|
payload = {
|
|
"mode": "apply" if args.apply else "dry-run",
|
|
"matched_count": plan.count,
|
|
"archived_count": archived_count,
|
|
"canonical_projects_preserved": sorted(CANONICAL_PROJECT_NAMES),
|
|
}
|
|
if args.show_names:
|
|
payload["matched_names"] = list(plan.names)
|
|
print(json.dumps(payload, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|