ffc4cdbd7d
The workshop + managed-global overlay surface fully covers the admin-SFTP flow that 'external' was a placeholder for. Drop the type from the model defaults, builder registry, routes, template, and tests, and add migration 0004 that deletes any leftover external rows along with their blueprint and job references. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
112 lines
3.3 KiB
Python
112 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import os
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from l4d2host.paths import get_left4me_root
|
|
|
|
from l4d2web.models import GlobalOverlaySource, Job, Overlay
|
|
from l4d2web.services.overlay_creation import generate_overlay_path
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManagedGlobalOverlay:
|
|
name: str
|
|
overlay_type: str
|
|
source_type: str
|
|
source_url: str
|
|
|
|
|
|
GLOBAL_OVERLAYS = (
|
|
ManagedGlobalOverlay(
|
|
name="l4d2center-maps",
|
|
overlay_type="l4d2center_maps",
|
|
source_type="l4d2center_csv",
|
|
source_url="https://l4d2center.com/maps/servers/index.csv",
|
|
),
|
|
ManagedGlobalOverlay(
|
|
name="cedapug-maps",
|
|
overlay_type="cedapug_maps",
|
|
source_type="cedapug_custom_page",
|
|
source_url="https://cedapug.com/custom",
|
|
),
|
|
)
|
|
|
|
MANAGED_GLOBAL_OVERLAY_TYPES = {overlay.overlay_type for overlay in GLOBAL_OVERLAYS}
|
|
USER_CREATABLE_TYPES = {"workshop"}
|
|
ADMIN_CREATABLE_TYPES = {"workshop"}
|
|
|
|
|
|
def is_creatable_overlay_type(overlay_type: str, *, admin: bool) -> bool:
|
|
allowed = ADMIN_CREATABLE_TYPES if admin else USER_CREATABLE_TYPES
|
|
return overlay_type in allowed
|
|
|
|
|
|
def ensure_global_overlays(session: Session) -> set[str]:
|
|
created_sources: set[str] = set()
|
|
for managed in GLOBAL_OVERLAYS:
|
|
overlay = session.scalar(
|
|
select(Overlay).where(Overlay.name == managed.name, Overlay.user_id.is_(None))
|
|
)
|
|
overlay_created = overlay is None
|
|
if overlay is None:
|
|
overlay = Overlay(name=managed.name, path="", type=managed.overlay_type, user_id=None)
|
|
session.add(overlay)
|
|
session.flush()
|
|
overlay.path = generate_overlay_path(overlay.id)
|
|
else:
|
|
overlay.type = managed.overlay_type
|
|
overlay.user_id = None
|
|
if not overlay.path:
|
|
overlay.path = generate_overlay_path(overlay.id)
|
|
|
|
target = get_left4me_root() / "overlays" / overlay.path
|
|
os.makedirs(target, exist_ok=not overlay_created)
|
|
|
|
source = session.scalar(
|
|
select(GlobalOverlaySource).where(GlobalOverlaySource.source_key == managed.name)
|
|
)
|
|
if source is None:
|
|
source = GlobalOverlaySource(
|
|
overlay_id=overlay.id,
|
|
source_key=managed.name,
|
|
source_type=managed.source_type,
|
|
source_url=managed.source_url,
|
|
)
|
|
session.add(source)
|
|
created_sources.add(managed.name)
|
|
else:
|
|
source.overlay_id = overlay.id
|
|
source.source_type = managed.source_type
|
|
source.source_url = managed.source_url
|
|
|
|
session.flush()
|
|
|
|
return created_sources
|
|
|
|
|
|
def enqueue_refresh_global_overlays(session: Session, *, user_id: int | None) -> Job:
|
|
existing = session.scalar(
|
|
select(Job)
|
|
.where(
|
|
Job.operation == "refresh_global_overlays",
|
|
Job.state.in_({"queued", "running", "cancelling"}),
|
|
)
|
|
.order_by(Job.created_at, Job.id)
|
|
)
|
|
if existing is not None:
|
|
return existing
|
|
|
|
job = Job(
|
|
user_id=user_id,
|
|
server_id=None,
|
|
overlay_id=None,
|
|
operation="refresh_global_overlays",
|
|
state="queued",
|
|
)
|
|
session.add(job)
|
|
session.flush()
|
|
return job
|