214 lines
8.2 KiB
Python
214 lines
8.2 KiB
Python
"""
|
|
===============================================================================
|
|
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]]:
|
|
init_manual_absences_table()
|
|
with get_connection(row_factory=True) as conn:
|
|
cursor = conn.cursor()
|
|
if absence_type:
|
|
cursor.execute("""
|
|
SELECT id, absence_type, fio, fio_clean, department, position, date_start, date_end, reason, comment, created_at
|
|
FROM manual_absences
|
|
WHERE absence_type = ?
|
|
ORDER BY id DESC
|
|
""", (absence_type.upper(),))
|
|
else:
|
|
cursor.execute("""
|
|
SELECT id, absence_type, fio, fio_clean, department, position, date_start, date_end, reason, comment, created_at
|
|
FROM manual_absences
|
|
ORDER BY id DESC
|
|
""")
|
|
return [dict(r) for r in cursor.fetchall()]
|
|
|
|
|
|
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()] |