53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
"""Persistent mapping between Telegram topics and Tracker projects."""
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MAP_FILE = Path(__file__).parent / "data" / "topic_map.json"
|
|
|
|
|
|
class TopicMap:
|
|
"""Bidirectional mapping: topic_id <-> project_uuid."""
|
|
|
|
def __init__(self):
|
|
self._by_topic: dict[int, str] = {} # topic_id -> project_uuid
|
|
self._by_project: dict[str, int] = {} # project_uuid -> topic_id
|
|
self._load()
|
|
|
|
def _load(self):
|
|
if MAP_FILE.exists():
|
|
data = json.loads(MAP_FILE.read_text())
|
|
for tid_str, puuid in data.items():
|
|
tid = int(tid_str)
|
|
self._by_topic[tid] = puuid
|
|
self._by_project[puuid] = tid
|
|
logger.info("Loaded %d topic mappings", len(self._by_topic))
|
|
|
|
def _save(self):
|
|
MAP_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
data = {str(tid): puuid for tid, puuid in self._by_topic.items()}
|
|
MAP_FILE.write_text(json.dumps(data, indent=2))
|
|
|
|
def set(self, topic_id: int, project_uuid: str):
|
|
self._by_topic[topic_id] = project_uuid
|
|
self._by_project[project_uuid] = topic_id
|
|
self._save()
|
|
|
|
def get_project(self, topic_id: int) -> str | None:
|
|
return self._by_topic.get(topic_id)
|
|
|
|
def get_topic(self, project_uuid: str) -> int | None:
|
|
return self._by_project.get(project_uuid)
|
|
|
|
def remove_by_topic(self, topic_id: int):
|
|
puuid = self._by_topic.pop(topic_id, None)
|
|
if puuid:
|
|
self._by_project.pop(puuid, None)
|
|
self._save()
|
|
|
|
def all(self) -> dict[int, str]:
|
|
return dict(self._by_topic)
|