""" =============================================================================== FILE: services/manual_absences_repo.py ROLE: Репозиторий ручных реестров ("Мест. командир.", "Иное") и поиск по штату 1С. =============================================================================== """ import os import csv from datetime import datetime from typing import List, Dict, Any, Optional from core.connection import get_connection from config import normalize_fio, DATA_DIR REASONS_CSV_PATH = os.path.join(DATA_DIR, "static_reason_absence.csv") def init_manual_absences_table() -> None: with get_connection() as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS manual_absences ( id INTEGER PRIMARY KEY AUTOINCREMENT, absence_type TEXT NOT NULL, -- 'LOCAL_TRIP' или 'OTHER' fio TEXT NOT NULL, fio_clean TEXT NOT NULL, department TEXT DEFAULT '', position TEXT DEFAULT '', date_start TEXT, -- 'YYYY-MM-DD' date_end TEXT, -- 'YYYY-MM-DD' reason TEXT NOT NULL, comment TEXT DEFAULT '', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) conn.execute("CREATE INDEX IF NOT EXISTS idx_manual_abs_dates ON manual_absences(date_start, date_end);") conn.execute("CREATE INDEX IF NOT EXISTS idx_manual_abs_fio ON manual_absences(fio_clean);") conn.commit() def get_static_reasons() -> List[str]: """Возвращает список причин из data/static_reason_absence.csv.""" if not os.path.exists(REASONS_CSV_PATH): # Если файл еще не создан, создаем базовый набор причин os.makedirs(os.path.dirname(REASONS_CSV_PATH), exist_ok=True) default_reasons = ["По семейным обстоятельствам", "Медосмотр", "Сдача крови", "Учебный отпуск", "Административный отпуск"] with open(REASONS_CSV_PATH, "w", encoding="utf-8", newline="") as f: writer = csv.writer(f) writer.writerow(["reason"]) for r in default_reasons: writer.writerow([r]) return default_reasons reasons = [] try: with open(REASONS_CSV_PATH, "r", encoding="utf-8") as f: reader = csv.reader(f) for row in reader: if row and row[0].strip() and row[0].strip().lower() != "reason": reasons.append(row[0].strip()) except Exception: pass return reasons def search_staff_suggestions(query: str, limit: int = 15) -> List[Dict[str, str]]: """Живой поиск сотрудников по zup_staff для автокомплита.""" q = (query or "").strip() if not q or len(q) < 2: return [] # Приводим к разным регистрам для гарантированного поиска кириллицы в SQLite q_lower = q.lower() q_title = q.capitalize() with get_connection(row_factory=True) as conn: cursor = conn.cursor() # 1. Находим действительно самый свежий срез штата (по created_at или по структуре даты ГГГГ-ММ-ДД) cursor.execute(""" SELECT snapshot_date FROM zup_staff ORDER BY SUBSTR(snapshot_date, 7, 4) DESC, SUBSTR(snapshot_date, 4, 2) DESC, SUBSTR(snapshot_date, 1, 2) DESC, id DESC LIMIT 1 """) row = cursor.fetchone() latest_date = row[0] if row else None if not latest_date: return [] # 2. Поиск с сортировкой: сначала те, у кого фамилия НАЧИНАЕТСЯ с запроса sql = """ SELECT DISTINCT fio, fio_clean, department, position FROM zup_staff WHERE snapshot_date = ? AND ( fio LIKE ? OR fio LIKE ? OR fio_clean LIKE ? OR fio_clean LIKE ? OR fio LIKE ? OR fio_clean LIKE ? ) ORDER BY CASE WHEN fio LIKE ? OR fio_clean LIKE ? THEN 0 ELSE 1 END, fio ASC LIMIT ? """ prefix_pattern_title = f"{q_title}%" prefix_pattern_lower = f"{q_lower}%" any_pattern_title = f"%{q_title}%" any_pattern_lower = f"%{q_lower}%" cursor.execute(sql, ( latest_date, prefix_pattern_title, prefix_pattern_lower, prefix_pattern_title, prefix_pattern_lower, any_pattern_title, any_pattern_lower, prefix_pattern_title, prefix_pattern_title, limit )) rows = cursor.fetchall() return [ { "fio": r["fio"], "fio_clean": r["fio_clean"], "department": r["department"] or "—", "position": r["position"] or "—" } for r in rows ] def add_manual_absence( absence_type: str, fio: str, reason: str, department: str = "", position: str = "", date_start: Optional[str] = None, date_end: Optional[str] = None, comment: str = "" ) -> int: init_manual_absences_table() clean_fio = normalize_fio(fio) if not clean_fio: return 0 today_str = datetime.now().strftime("%Y-%m-%d") d_start = date_start.strip() if date_start and date_start.strip() else today_str d_end = date_end.strip() if date_end and date_end.strip() else today_str with get_connection() as conn: cursor = conn.cursor() cursor.execute(""" INSERT INTO manual_absences ( absence_type, fio, fio_clean, department, position, date_start, date_end, reason, comment ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, (absence_type.upper(), fio.strip(), clean_fio, department.strip(), position.strip(), d_start, d_end, reason.strip(), comment.strip())) conn.commit() return cursor.lastrowid def delete_manual_absence(item_id: int) -> bool: init_manual_absences_table() with get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM manual_absences WHERE id = ?", (item_id,)) conn.commit() return cursor.rowcount > 0 def get_manual_absences_list(absence_type: Optional[str] = None) -> List[Dict[str, Any]]: today_str = datetime.now().strftime("%Y-%m-%d") # ⭐️ Обязательно row_factory=True для преобразования строк SQLite в dict with get_connection(row_factory=True) as conn: cursor = conn.cursor() # 1. Автоочистка просроченных командировок и отсутствий cursor.execute(""" DELETE FROM manual_absences WHERE date_end IS NOT NULL AND date_end != '' AND date_end < ? """, (today_str,)) conn.commit() # 2. Выборка активных записей query = "SELECT id, absence_type, fio, department, position, reason, date_start, date_end, comment FROM manual_absences" params = [] if absence_type: query += " WHERE absence_type = ?" params.append(absence_type) query += " ORDER BY id DESC" cursor.execute(query, params) rows = cursor.fetchall() return [dict(r) for r in rows] def get_active_manual_absences_for_date(date_str: str) -> List[Dict[str, Any]]: """Выбирает записи, активные на дату отчета (формат даты ДД.ММ.ГГГГ).""" init_manual_absences_table() try: dt_target = datetime.strptime(date_str.replace('_', '.'), "%d.%m.%Y").strftime("%Y-%m-%d") except Exception: dt_target = datetime.now().strftime("%Y-%m-%d") with get_connection(row_factory=True) as conn: cursor = conn.cursor() cursor.execute(""" SELECT id, absence_type, fio, fio_clean, department, position, date_start, date_end, reason FROM manual_absences WHERE (date_start IS NULL OR date_start <= ?) AND (date_end IS NULL OR date_end >= ?) """, (dt_target, dt_target)) return [dict(r) for r in cursor.fetchall()]