"""Соединение с SQLite и схема хаба. Схема создаётся при первом подключении; версия — в setting.schema_version.""" import itertools import json import sqlite3 from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path from . import backup, paths SCHEMA_VERSION = 5 SCHEMA = """ CREATE TABLE IF NOT EXISTS setting ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS issue ( key TEXT PRIMARY KEY, id TEXT, issue_type TEXT, summary TEXT, description TEXT, status TEXT, priority TEXT, assignee TEXT, reporter TEXT, duedate TEXT, planned_start TEXT, planned_end TEXT, epic_key TEXT, labels TEXT NOT NULL DEFAULT '[]', sprint TEXT, created TEXT, jira_updated TEXT, fetched_at TEXT NOT NULL, neighbor INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS issue_link ( source_key TEXT NOT NULL, target_key TEXT NOT NULL, link_type TEXT NOT NULL, PRIMARY KEY (source_key, target_key, link_type) ); CREATE INDEX IF NOT EXISTS ix_issue_link_target ON issue_link(target_key); CREATE TABLE IF NOT EXISTS issue_comment ( key TEXT NOT NULL, comment_id TEXT NOT NULL, author TEXT, created TEXT, updated TEXT, body TEXT NOT NULL DEFAULT '', created_at TEXT, PRIMARY KEY (key, comment_id) ); CREATE TABLE IF NOT EXISTS issue_change ( history_id TEXT NOT NULL, seq INTEGER NOT NULL, key TEXT NOT NULL, at TEXT, created TEXT NOT NULL, author TEXT, field TEXT NOT NULL, field_id TEXT, field_name TEXT NOT NULL, from_value TEXT, from_string TEXT, to_value TEXT, to_string TEXT, fetched_at TEXT NOT NULL, PRIMARY KEY (history_id, seq) ); CREATE INDEX IF NOT EXISTS ix_issue_change_at ON issue_change(at); CREATE INDEX IF NOT EXISTS ix_issue_change_key ON issue_change(key, at); CREATE TABLE IF NOT EXISTS issue_note ( key TEXT PRIMARY KEY, fact_status TEXT, work_kind TEXT, done_text TEXT, rest_text TEXT, owner_note TEXT, updated_at TEXT, updated_by TEXT ); CREATE TABLE IF NOT EXISTS batch ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, author TEXT NOT NULL, created_at TEXT NOT NULL, cancelled_at TEXT, estimate_seconds INTEGER NOT NULL DEFAULT 0, project TEXT ); CREATE TABLE IF NOT EXISTS command ( id INTEGER PRIMARY KEY AUTOINCREMENT, batch_id INTEGER NOT NULL REFERENCES batch(id), seq INTEGER NOT NULL, type TEXT NOT NULL, params TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'queued', claimed_at TEXT, deliveries INTEGER NOT NULL DEFAULT 0, finished_at TEXT, result TEXT, error TEXT, jira_calls INTEGER NOT NULL DEFAULT 0, rate_limited INTEGER NOT NULL DEFAULT 0, pause_seconds INTEGER, depends_on INTEGER REFERENCES command(id) ); CREATE INDEX IF NOT EXISTS ix_command_status ON command(status, batch_id, seq); CREATE TABLE IF NOT EXISTS sync_run ( id INTEGER PRIMARY KEY AUTOINCREMENT, started_at TEXT NOT NULL, finished_at TEXT, kind TEXT NOT NULL, query TEXT, issues_count INTEGER NOT NULL DEFAULT 0, links_count INTEGER NOT NULL DEFAULT 0, jira_calls INTEGER NOT NULL DEFAULT 0, rate_limited INTEGER NOT NULL DEFAULT 0, command_id INTEGER, history_at TEXT ); CREATE TABLE IF NOT EXISTS project ( label TEXT PRIMARY KEY, name TEXT NOT NULL, board_id INTEGER, board_type TEXT, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS agent_status ( id INTEGER PRIMARY KEY CHECK (id = 1), last_seen_at TEXT, agent_version TEXT, settings_version INTEGER, current_command_id INTEGER, last_jira_call_at TEXT ); """ def _migrate_projects(conn) -> None: from . import projects # внутри функции: projects импортирует db projects.migrate_v3(conn) def _add_column(table: str, column: str): """Шаг миграции: добавить колонку, если её нет. Базы v1/v2 получают issue_comment из SCHEMA уже с новой колонкой.""" def step(conn) -> None: if column not in {r["name"] for r in conn.execute(f"PRAGMA table_info({table})")}: conn.execute(f"ALTER TABLE {table} ADD COLUMN {column} TEXT") return step def _fill_comment_utc(conn) -> None: for r in conn.execute("SELECT key, comment_id, created FROM issue_comment").fetchall(): conn.execute("UPDATE issue_comment SET created_at = ? WHERE key = ? AND comment_id = ?", (jira_utc(r["created"]), r["key"], r["comment_id"])) # Шаги миграции: версия → SQL, переводящий схему в версию + 1 (design.md bridge-reliability, решение 5) MIGRATIONS = { 1: ["ALTER TABLE command ADD COLUMN deliveries INTEGER NOT NULL DEFAULT 0", "UPDATE command SET deliveries = 1 WHERE status = 'claimed'"], # уже выданные — одна выдача 2: [], # v2 → v3: issue_comment создаётся SCHEMA (CREATE TABLE IF NOT EXISTS), данных переносить не нужно 3: ["ALTER TABLE batch ADD COLUMN project TEXT", _migrate_projects], # v3 → v4: проекты-метки 4: [_add_column("issue_comment", "created_at"), _add_column("sync_run", "history_at"), # v4 → v5: журнал изменений _fill_comment_utc], } def now() -> str: """Текущее время UTC в ISO-8601 с секундами — единый формат дат в базе.""" return datetime.now(timezone.utc).replace(microsecond=0).isoformat() def connect(path) -> sqlite3.Connection: """Открыть базу (создать при отсутствии), применить схему. Соединение в автокоммите; для атомарных операций — transaction().""" is_memory = str(path) == ":memory:" if not is_memory and Path(path).parent != Path("."): # путь без каталога — рабочий каталог процесса не трогаем paths.secure_dir(Path(path).parent) conn = sqlite3.connect(str(path), timeout=10, isolation_level=None, check_same_thread=False) conn.row_factory = sqlite3.Row if not is_memory: conn.execute("PRAGMA journal_mode = WAL") conn.execute("PRAGMA foreign_keys = ON") migrate(conn, None if is_memory else path) return conn def _schema_version(conn) -> int | None: row = conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone() return None if row is None else int(json.loads(row["value"])) def migrate(conn: sqlite3.Connection, path=None) -> None: """Создать недостающие таблицы и поднять схему существующей базы до SCHEMA_VERSION. Копия базы (<каталог базы>/backups, без ротации) и миграция — под одной блокировкой записи (BEGIN IMMEDIATE): параллельный connect дождётся её, перечитает версию и не станет ни копировать, ни мигрировать повторно (M-2 #12). Копию делает отдельное соединение: VACUUM INTO внутри транзакции SQLite запрещает, а читатель при WAL видит базу до миграции. Вызывается только вне транзакции.""" conn.executescript(SCHEMA) version = _schema_version(conn) if version is None: conn.execute("INSERT OR IGNORE INTO setting(key, value) VALUES ('schema_version', ?)", (json.dumps(SCHEMA_VERSION),)) elif version < SCHEMA_VERSION: with transaction(conn): current = _schema_version(conn) if current < SCHEMA_VERSION: if path is not None: reader = sqlite3.connect(str(path), timeout=10) try: backup.make_backup(reader, Path(path).parent / "backups", f"schema-v{current}", keep=0) finally: reader.close() while current < SCHEMA_VERSION: for step in MIGRATIONS[current]: step(conn) if callable(step) else conn.execute(step) current += 1 conn.execute("UPDATE setting SET value = ? WHERE key = 'schema_version'", (json.dumps(current),)) conn.execute("INSERT OR IGNORE INTO agent_status(id) VALUES (1)") _savepoints = itertools.count(1) @contextmanager def transaction(conn: sqlite3.Connection): """BEGIN IMMEDIATE … COMMIT; при исключении — ROLLBACK. Нужна там, где читаем и пишем вместе (claim, отчёт). Вложенный вызов (транзакция уже открыта) идёт через SAVEPOINT: при успехе RELEASE, при исключении ROLLBACK TO + RELEASE и исключение пробрасывается — вызывающий может его перехватить, внешняя транзакция при этом остаётся целой, а записи вложенного блока не сохраняются. Соединение не должно делиться между потоками: каждый поток/запрос открывает своё.""" if conn.in_transaction: name = f"sp_{next(_savepoints)}" conn.execute(f"SAVEPOINT {name}") try: yield conn except BaseException: conn.execute(f"ROLLBACK TO {name}") conn.execute(f"RELEASE {name}") raise else: conn.execute(f"RELEASE {name}") return conn.execute("BEGIN IMMEDIATE") try: yield conn except BaseException: conn.execute("ROLLBACK") raise else: conn.execute("COMMIT") def jira_utc(value) -> str | None: """Время Jira (`2026-10-22T13:05:11.000+0300`, `+03:00`, `Z`, с дробной частью или без) → ISO-8601 UTC в формате now(); без смещения — UTC; неразбираемое — None.""" if not isinstance(value, str) or not value: return None s = value[:-1] + "+00:00" if value.endswith("Z") else value if len(s) > 5 and s[-5] in "+-" and s[-4:].isdigit(): # ±HHMM → ±HH:MM s = s[:-2] + ":" + s[-2:] try: d = datetime.fromisoformat(s) except ValueError: return None d = d if d.tzinfo else d.replace(tzinfo=timezone.utc) return d.astimezone(timezone.utc).replace(microsecond=0).isoformat() def utcnow() -> datetime: return datetime.now(timezone.utc).replace(microsecond=0) def parse(ts: str) -> datetime: """ISO-8601 из базы → aware datetime (naive считаем UTC).""" d = datetime.fromisoformat(ts) return d if d.tzinfo else d.replace(tzinfo=timezone.utc)