lokses
0da9c13192
Иерархия рамочных/дочерних договоров и платежей, интернет-биллинг, месячные IT-отчёты (импорт, DOCX/PDF, шаблоны выводов). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
339 lines
13 KiB
Python
339 lines
13 KiB
Python
"""Месячный IT-отчёт из WorkLog (логика как в report-tool)."""
|
||
from __future__ import annotations
|
||
|
||
import calendar
|
||
import re
|
||
from collections import Counter, defaultdict
|
||
from datetime import date
|
||
|
||
from django.contrib.auth import get_user_model
|
||
from django.utils import timezone
|
||
|
||
from apps.it_manager.models import WorkLog
|
||
from apps.it_manager.report_import import default_report_tool_root, load_company, normalize_name
|
||
from apps.locations.models.base import pharmacy_object_code
|
||
from apps.reports.it_monthly.conclusions import build_conclusions
|
||
|
||
Employee = get_user_model()
|
||
|
||
MONTHS_RU = [
|
||
'', 'Январь', 'Февраль', 'Март', 'Апрель', 'Май', 'Июнь',
|
||
'Июль', 'Август', 'Сентябрь', 'Октябрь', 'Ноябрь', 'Декабрь',
|
||
]
|
||
|
||
INFRA_DIRECTION = 'Администрирование серверов, БД, мониторинг и безопасность'
|
||
SYSTEM_DESC_RE = re.compile(r'^([^:]{3,80}):\s+(.+)$', re.DOTALL)
|
||
# Коды аптек по юрлицам: А12, П10, Т5, Г3, Л1 (лат./кир.)
|
||
_PHARMACY_OBJECT_RE = re.compile(r'^[АAПPТTГGЛL]\d+$', re.I)
|
||
_OBJECT_LAT_TO_CYR = str.maketrans('APTGL', 'АПТГЛ')
|
||
|
||
|
||
def _is_pharmacy_object_code(code: str) -> bool:
|
||
raw = (code or '').strip().upper().translate(_OBJECT_LAT_TO_CYR)
|
||
return bool(_PHARMACY_OBJECT_RE.match(raw))
|
||
|
||
|
||
def split_work_items(text: str) -> list[str]:
|
||
"""Длинные журналы вида «а; б; в» → отдельные пункты для читаемого UI."""
|
||
text = re.sub(r'\s+', ' ', (text or '').strip())
|
||
if not text:
|
||
return []
|
||
parts = [p.strip(' .;') for p in re.split(r'\s*;\s*', text) if p.strip(' .;')]
|
||
if len(parts) <= 1:
|
||
return [text]
|
||
return [p[0].upper() + p[1:] if p else p for p in parts]
|
||
|
||
|
||
def department_head(department_id) -> Employee | None:
|
||
"""Керівник відділу — та сама евристика, що на /structure/."""
|
||
if not department_id:
|
||
return None
|
||
emps = list(
|
||
Employee.objects.filter(
|
||
is_active=True,
|
||
department_id=department_id,
|
||
user_location_type='office',
|
||
)
|
||
)
|
||
ids = {e.pk for e in emps}
|
||
managers_in_dept = {e.manager_id for e in emps if e.manager_id in ids}
|
||
candidates = [e for e in emps if e.pk in managers_in_dept]
|
||
if not candidates:
|
||
return None
|
||
return next((e for e in candidates if e.manager_id not in ids), None) or candidates[0]
|
||
|
||
|
||
def user_can_view_it_monthly_report(user) -> bool:
|
||
"""Месячный отчёт — только руководителю отдела (или superuser)."""
|
||
if not getattr(user, 'is_authenticated', False):
|
||
return False
|
||
if getattr(user, 'is_superuser', False):
|
||
return True
|
||
head = department_head(getattr(user, 'department_id', None))
|
||
return head is not None and head.pk == user.pk
|
||
|
||
|
||
def parse_month_id(value: str | None) -> tuple[int, int]:
|
||
today = timezone.localdate()
|
||
if not value:
|
||
return today.year, today.month
|
||
m = re.fullmatch(r'(\d{4})-(\d{2})', (value or '').strip())
|
||
if not m:
|
||
return today.year, today.month
|
||
year, month = int(m.group(1)), int(m.group(2))
|
||
if not (1 <= month <= 12):
|
||
return today.year, today.month
|
||
return year, month
|
||
|
||
|
||
def month_id(year: int, month: int) -> str:
|
||
return f'{year:04d}-{month:02d}'
|
||
|
||
|
||
def available_report_months(limit: int = 24) -> list[str]:
|
||
"""YYYY-MM из журнала + текущий месяц, новые сверху."""
|
||
today = timezone.localdate()
|
||
months = {month_id(today.year, today.month)}
|
||
for row in WorkLog.objects.dates('date', 'month', order='DESC')[:limit]:
|
||
months.add(month_id(row.year, row.month))
|
||
return sorted(months, reverse=True)
|
||
|
||
|
||
def location_object_code(location) -> str:
|
||
if location is None:
|
||
return ''
|
||
if location.location_type == 'office':
|
||
return 'ОФИС'
|
||
if location.location_type == 'warehouse':
|
||
return 'Склад'
|
||
pharmacy = getattr(location, 'pharmacy', None)
|
||
if pharmacy is not None:
|
||
return pharmacy_object_code(pharmacy)
|
||
return ''
|
||
|
||
|
||
def classify_direction(text: str, directions: list[dict]) -> str:
|
||
low = (text or '').lower()
|
||
for direction in directions:
|
||
for kw in direction.get('keywords') or []:
|
||
if kw.lower() in low:
|
||
return direction.get('name') or direction.get('id') or 'Прочие работы'
|
||
return 'Прочие работы'
|
||
|
||
|
||
def _staff_order(company: dict) -> list[dict]:
|
||
return list(company.get('staff') or [])
|
||
|
||
|
||
def _match_staff_id(employee, company: dict) -> str | None:
|
||
last = normalize_name(employee.last_name)
|
||
full = normalize_name(employee.get_full_name() or '')
|
||
for staff in _staff_order(company):
|
||
for token in staff.get('match') or []:
|
||
if normalize_name(token) and normalize_name(token) in full:
|
||
return staff['id']
|
||
name_last = normalize_name((staff.get('name') or '').split()[0] if staff.get('name') else '')
|
||
if name_last and name_last == last:
|
||
return staff['id']
|
||
return None
|
||
|
||
|
||
def build_monthly_report(year: int, month: int, *, company: dict | None = None) -> dict:
|
||
if company is None:
|
||
root = default_report_tool_root()
|
||
company = load_company(root) if (root / 'company.json').exists() else {'staff': [], 'directions': []}
|
||
|
||
directions = company.get('directions') or []
|
||
start = date(year, month, 1)
|
||
end = date(year, month, calendar.monthrange(year, month)[1])
|
||
|
||
logs = list(
|
||
WorkLog.objects.filter(date__gte=start, date__lte=end)
|
||
.select_related('technician', 'location', 'work_type', 'ticket')
|
||
.prefetch_related('location__pharmacy__organization')
|
||
.order_by('date', 'id')
|
||
)
|
||
|
||
by_tech: dict[int, list] = defaultdict(list)
|
||
for log in logs:
|
||
by_tech[log.technician_id].append(log)
|
||
|
||
direction_counts: Counter[str] = Counter()
|
||
direction_people: dict[str, set[str]] = defaultdict(set)
|
||
objects: list[str] = []
|
||
seen_obj: set[str] = set()
|
||
|
||
# упорядочить сотрудников: сначала staff из company, потом остальные
|
||
staff_by_id = {s['id']: s for s in _staff_order(company)}
|
||
tech_to_staff: dict[int, str | None] = {}
|
||
for tech_id, tech_logs in by_tech.items():
|
||
tech_to_staff[tech_id] = _match_staff_id(tech_logs[0].technician, company)
|
||
|
||
ordered_tech_ids: list[int] = []
|
||
used: set[int] = set()
|
||
for staff in _staff_order(company):
|
||
for tech_id, sid in tech_to_staff.items():
|
||
if sid == staff['id'] and tech_id not in used:
|
||
ordered_tech_ids.append(tech_id)
|
||
used.add(tech_id)
|
||
break
|
||
for tech_id in by_tech:
|
||
if tech_id not in used:
|
||
ordered_tech_ids.append(tech_id)
|
||
|
||
employees_out = []
|
||
people_active = 0
|
||
for tech_id in ordered_tech_ids:
|
||
tech_logs = by_tech[tech_id]
|
||
tech = tech_logs[0].technician
|
||
staff_id = tech_to_staff.get(tech_id)
|
||
staff_meta = staff_by_id.get(staff_id) if staff_id else None
|
||
name = (staff_meta or {}).get('name') or tech.get_full_name() or tech.username
|
||
short = (staff_meta or {}).get('short') or name
|
||
role = (staff_meta or {}).get('role') or ''
|
||
tasks = []
|
||
systems = []
|
||
for log in tech_logs:
|
||
obj = location_object_code(log.location)
|
||
if not obj:
|
||
# префикс «А12 — …» из импорта без локации
|
||
m = re.match(r'^([АAПPТTГGЛL]\d+|ОФИС|Склад)\s*[—\-]\s*(.+)$', log.description, re.I | re.S)
|
||
if m:
|
||
obj = m.group(1).upper().replace('A', 'А')
|
||
work = m.group(2).strip()
|
||
else:
|
||
work = log.description
|
||
else:
|
||
work = log.description
|
||
|
||
sys_m = SYSTEM_DESC_RE.match(work)
|
||
if sys_m and (log.result or '').startswith('import:report-tool:'):
|
||
systems.append({'name': sys_m.group(1).strip(), 'work': sys_m.group(2).strip()})
|
||
direction_counts[INFRA_DIRECTION] += 1
|
||
direction_people[INFRA_DIRECTION].add(short)
|
||
else:
|
||
tasks.append({
|
||
'date': log.date.strftime('%d.%m.%Y'),
|
||
'object': obj,
|
||
'work': work,
|
||
'work_items': split_work_items(work),
|
||
'work_type': log.work_type.name if log.work_type_id else '',
|
||
'ticket': log.ticket.number if log.ticket_id else '',
|
||
'hours': float(log.hours_spent) if log.hours_spent is not None else None,
|
||
})
|
||
label = classify_direction(f'{obj} {work}', directions)
|
||
direction_counts[label] += 1
|
||
direction_people[label].add(short)
|
||
|
||
for code in filter(None, [obj]):
|
||
key = code.upper()
|
||
if key not in seen_obj:
|
||
seen_obj.add(key)
|
||
objects.append(code)
|
||
|
||
if tasks or systems:
|
||
people_active += 1
|
||
employees_out.append({
|
||
'id': staff_id or f'user-{tech_id}',
|
||
'name': name,
|
||
'short': short,
|
||
'role': role,
|
||
'tasks': tasks,
|
||
'systems': systems,
|
||
'tasks_count': len(tasks),
|
||
'systems_count': len(systems),
|
||
})
|
||
|
||
summary = [
|
||
{
|
||
'name': name,
|
||
'count': count,
|
||
'people': ', '.join(sorted(direction_people.get(name, []))),
|
||
}
|
||
for name, count in direction_counts.most_common()
|
||
]
|
||
|
||
pharmacies = [o for o in objects if _is_pharmacy_object_code(o)]
|
||
# Офис, склад и прочие точки без кода аптеки — в «Другие объекты»
|
||
others = [o for o in objects if not _is_pharmacy_object_code(o)]
|
||
|
||
report = {
|
||
'id': month_id(year, month),
|
||
'year': year,
|
||
'month': month,
|
||
'title': f'{MONTHS_RU[month]} {year} года',
|
||
'period_from': start.strftime('%d.%m.%Y'),
|
||
'period_to': end.strftime('%d.%m.%Y'),
|
||
'total_tasks': sum(e['tasks_count'] for e in employees_out),
|
||
'people_active': people_active,
|
||
'objects_count': len(objects),
|
||
'objects': objects,
|
||
'pharmacies': pharmacies,
|
||
'other_points': others,
|
||
'extra_points': [],
|
||
'summary': summary,
|
||
'employees': employees_out,
|
||
'available_months': available_report_months(),
|
||
}
|
||
report['conclusions'] = build_conclusions(report, company)
|
||
return report
|
||
|
||
|
||
def report_to_markdown(report: dict) -> str:
|
||
lines = [
|
||
f"# IT-отчёт — {report['title']}",
|
||
'',
|
||
f"**Период:** {report['period_from']} – {report['period_to']}",
|
||
f"**Всего работ:** {report['total_tasks']}",
|
||
f"**Активных сотрудников:** {report['people_active']}",
|
||
f"**Объектов:** {report['objects_count']}",
|
||
'',
|
||
'## Сводка по направлениям',
|
||
'',
|
||
]
|
||
if report['summary']:
|
||
for row in report['summary']:
|
||
people = f" ({row['people']})" if row['people'] else ''
|
||
lines.append(f"- **{row['name']}** — {row['count']}{people}")
|
||
else:
|
||
lines.append('_Нет записей за месяц._')
|
||
|
||
if report['pharmacies'] or report['other_points']:
|
||
lines.extend(['', '## Объекты', ''])
|
||
if report['pharmacies']:
|
||
lines.append(f"- Аптеки: {', '.join(report['pharmacies'])}.")
|
||
if report['other_points']:
|
||
lines.append(f"- Другие объекты: {', '.join(report['other_points'])}.")
|
||
|
||
for emp in report['employees']:
|
||
if not emp['tasks'] and not emp['systems']:
|
||
continue
|
||
lines.extend(['', f"## {emp['name']}", ''])
|
||
if emp.get('role'):
|
||
lines.append(f"*{emp['role']}*")
|
||
lines.append('')
|
||
for sys in emp['systems']:
|
||
lines.append(f"### {sys['name']}")
|
||
lines.append(sys['work'])
|
||
lines.append('')
|
||
for task in emp['tasks']:
|
||
obj = f" {task['object']}" if task['object'] else ''
|
||
items = task.get('work_items') or [task['work']]
|
||
if len(items) == 1:
|
||
lines.append(f"- {task['date']}{obj} — {items[0]}")
|
||
else:
|
||
lines.append(f"- {task['date']}{obj}:")
|
||
for item in items:
|
||
lines.append(f" - {item}")
|
||
|
||
if report.get('conclusions'):
|
||
lines.extend(['', '## Основные выводы и рекомендации', ''])
|
||
for item in report['conclusions']:
|
||
lines.append(item.get('text') or '')
|
||
for bullet in item.get('bullets') or []:
|
||
lines.append(f' - {bullet}')
|
||
lines.append('')
|
||
|
||
lines.append('')
|
||
return '\n'.join(lines)
|