551 lines
20 KiB
Python
551 lines
20 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import logging
|
||
|
|
import re
|
||
|
|
from dataclasses import asdict, dataclass
|
||
|
|
from datetime import date, timedelta
|
||
|
|
from typing import Any
|
||
|
|
from urllib.parse import parse_qs, urlparse
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
|
||
|
|
from config import settings
|
||
|
|
from services.secret_box import decrypt_secret
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
DEFAULT_LIST_URL = "http://tw-compbase.supermicro.com:6699"
|
||
|
|
DEFAULT_FILL_URL = "http://tw-compbase.supermicro.com:6671"
|
||
|
|
|
||
|
|
_HIDDEN_FIELDS = (
|
||
|
|
"__VIEWSTATE",
|
||
|
|
"__VIEWSTATEGENERATOR",
|
||
|
|
"__EVENTVALIDATION",
|
||
|
|
"__VIEWSTATEENCRYPTED",
|
||
|
|
"__EVENTTARGET",
|
||
|
|
"__EVENTARGUMENT",
|
||
|
|
"__LASTFOCUS",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class AbnormalRow:
|
||
|
|
date: date
|
||
|
|
rid: str | None
|
||
|
|
exp_in: str
|
||
|
|
exp_out: str
|
||
|
|
actual_in: str
|
||
|
|
actual_out: str
|
||
|
|
status: str # missing_out | missing_in | other
|
||
|
|
|
||
|
|
def to_dict(self) -> dict[str, Any]:
|
||
|
|
data = asdict(self)
|
||
|
|
data["date"] = self.date.isoformat()
|
||
|
|
return data
|
||
|
|
|
||
|
|
|
||
|
|
class CompBaseClient:
|
||
|
|
def __init__(
|
||
|
|
self,
|
||
|
|
*,
|
||
|
|
list_url: str | None = None,
|
||
|
|
fill_url: str | None = None,
|
||
|
|
) -> None:
|
||
|
|
self.list_url = (list_url or getattr(settings, "compbase_list_url", "") or DEFAULT_LIST_URL).rstrip("/")
|
||
|
|
self.fill_url = (fill_url or getattr(settings, "compbase_fill_url", "") or DEFAULT_FILL_URL).rstrip("/")
|
||
|
|
|
||
|
|
def _credentials(self) -> tuple[str, str]:
|
||
|
|
username = (getattr(settings, "compbase_username", "") or "").strip()
|
||
|
|
encrypted = getattr(settings, "compbase_password", "") or ""
|
||
|
|
if not username or not encrypted:
|
||
|
|
username = (getattr(settings, "pts_username", "") or "").strip()
|
||
|
|
encrypted = getattr(settings, "pts_password", "") or ""
|
||
|
|
if not username or not encrypted:
|
||
|
|
raise ValueError(
|
||
|
|
"Missing CompBase credentials. Set CompBase or PTS username/password in Settings."
|
||
|
|
)
|
||
|
|
try:
|
||
|
|
password = decrypt_secret(encrypted)
|
||
|
|
except ValueError as exc:
|
||
|
|
raise ValueError("Stored password could not be decrypted. Re-enter password in Settings.") from exc
|
||
|
|
if not password:
|
||
|
|
raise ValueError("CompBase password is empty. Re-enter password in Settings.")
|
||
|
|
return username, password
|
||
|
|
|
||
|
|
def _auth(self):
|
||
|
|
try:
|
||
|
|
from httpx_ntlm import HttpNtlmAuth
|
||
|
|
except ImportError as exc:
|
||
|
|
raise ValueError("httpx-ntlm is not installed. Run: pip install httpx-ntlm") from exc
|
||
|
|
username, password = self._credentials()
|
||
|
|
return HttpNtlmAuth(username, password)
|
||
|
|
|
||
|
|
def _make_client(self) -> httpx.AsyncClient:
|
||
|
|
return httpx.AsyncClient(
|
||
|
|
auth=self._auth(),
|
||
|
|
timeout=60.0,
|
||
|
|
follow_redirects=True,
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _extract_hidden(html: str) -> dict[str, str]:
|
||
|
|
out: dict[str, str] = {}
|
||
|
|
for name in _HIDDEN_FIELDS:
|
||
|
|
m = re.search(
|
||
|
|
rf'<input[^>]+name=["\']{re.escape(name)}["\'][^>]*value=["\']([^"\']*)["\']',
|
||
|
|
html,
|
||
|
|
re.I,
|
||
|
|
)
|
||
|
|
if not m:
|
||
|
|
m = re.search(
|
||
|
|
rf'<input[^>]+id=["\']{re.escape(name)}["\'][^>]*value=["\']([^"\']*)["\']',
|
||
|
|
html,
|
||
|
|
re.I,
|
||
|
|
)
|
||
|
|
if m:
|
||
|
|
out[name] = m.group(1)
|
||
|
|
return out
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _extract_selected_selects(html: str) -> dict[str, str]:
|
||
|
|
out: dict[str, str] = {}
|
||
|
|
for m in re.finditer(
|
||
|
|
r'<select[^>]*name=["\']([^"\']+)["\'][^>]*>([\s\S]*?)</select>',
|
||
|
|
html,
|
||
|
|
re.I,
|
||
|
|
):
|
||
|
|
name, body = m.group(1), m.group(2)
|
||
|
|
selected = re.search(
|
||
|
|
r'<option[^>]*selected[^>]*(?:value=["\']([^"\']*)["\'])?[^>]*>([^<]*)',
|
||
|
|
body,
|
||
|
|
re.I,
|
||
|
|
)
|
||
|
|
if selected:
|
||
|
|
val = selected.group(1)
|
||
|
|
out[name] = val if val is not None else selected.group(2).strip()
|
||
|
|
continue
|
||
|
|
first = re.search(r'<option[^>]*value=["\']([^"\']*)["\']', body, re.I)
|
||
|
|
if first:
|
||
|
|
out[name] = first.group(1)
|
||
|
|
return out
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _extract_checked_radios(html: str) -> dict[str, str]:
|
||
|
|
out: dict[str, str] = {}
|
||
|
|
for m in re.finditer(r"<input([^>]+)/?>", html, re.I):
|
||
|
|
attrs = m.group(1)
|
||
|
|
typ = re.search(r'type=["\']([^"\']+)["\']', attrs, re.I)
|
||
|
|
if not typ or typ.group(1).lower() != "radio":
|
||
|
|
continue
|
||
|
|
if not re.search(r"\bchecked\b", attrs, re.I):
|
||
|
|
continue
|
||
|
|
nm = re.search(r'name=["\']([^"\']+)["\']', attrs, re.I)
|
||
|
|
val = re.search(r'value=["\']([^"\']*)["\']', attrs, re.I)
|
||
|
|
if nm:
|
||
|
|
out[nm.group(1)] = val.group(1) if val else "on"
|
||
|
|
return out
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _parse_us_date(text: str) -> date | None:
|
||
|
|
text = text.strip()
|
||
|
|
m = re.match(r"^(\d{1,2})/(\d{1,2})/(\d{4})$", text)
|
||
|
|
if not m:
|
||
|
|
return None
|
||
|
|
month, day, year = int(m.group(1)), int(m.group(2)), int(m.group(3))
|
||
|
|
try:
|
||
|
|
return date(year, month, day)
|
||
|
|
except ValueError:
|
||
|
|
return None
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _rid_from_href(href: str) -> str | None:
|
||
|
|
if not href:
|
||
|
|
return None
|
||
|
|
qs = parse_qs(urlparse(href).query)
|
||
|
|
rid_list = qs.get("RID") or qs.get("rid")
|
||
|
|
if rid_list and rid_list[0]:
|
||
|
|
return str(rid_list[0]).strip()
|
||
|
|
m = re.search(r"RID=(\d+)", href, re.I)
|
||
|
|
return m.group(1) if m else None
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _cell_text(cell_html: str) -> str:
|
||
|
|
text = re.sub(r"<[^>]+>", " ", cell_html)
|
||
|
|
text = re.sub(r" ", " ", text, flags=re.I)
|
||
|
|
return re.sub(r"\s+", " ", text).strip()
|
||
|
|
|
||
|
|
def _parse_table2(self, html: str) -> list[AbnormalRow]:
|
||
|
|
m = re.search(r'<table[^>]*id=["\']Table2["\'][^>]*>([\s\S]*?)</table>', html, re.I)
|
||
|
|
if not m:
|
||
|
|
return []
|
||
|
|
body = m.group(1)
|
||
|
|
rows: list[AbnormalRow] = []
|
||
|
|
for tr in re.finditer(r"<tr[^>]*>([\s\S]*?)</tr>", body, re.I):
|
||
|
|
cells = re.findall(r"<t[dh][^>]*>([\s\S]*?)</t[dh]>", tr.group(1), re.I)
|
||
|
|
if len(cells) < 5:
|
||
|
|
continue
|
||
|
|
day = self._parse_us_date(self._cell_text(cells[0]))
|
||
|
|
if day is None:
|
||
|
|
continue
|
||
|
|
# Layout: 日期 | 應上班 | 應下班 | 實上班 | 實下班 | [異常上班] | [異常下班]
|
||
|
|
exp_in = self._cell_text(cells[1]) if len(cells) > 1 else ""
|
||
|
|
exp_out = self._cell_text(cells[2]) if len(cells) > 2 else ""
|
||
|
|
actual_in = self._cell_text(cells[3]) if len(cells) > 3 else ""
|
||
|
|
actual_out = self._cell_text(cells[4]) if len(cells) > 4 else ""
|
||
|
|
abn_in_html = cells[5] if len(cells) > 5 else ""
|
||
|
|
abn_out_html = cells[6] if len(cells) > 6 else ""
|
||
|
|
|
||
|
|
rid: str | None = None
|
||
|
|
status = "ok"
|
||
|
|
href_m = re.search(r'href=["\']([^"\']+)["\']', abn_out_html, re.I)
|
||
|
|
if href_m:
|
||
|
|
rid = self._rid_from_href(href_m.group(1))
|
||
|
|
if "應刷未刷" in abn_out_html or rid:
|
||
|
|
status = "missing_out"
|
||
|
|
elif "應刷未刷" in abn_in_html:
|
||
|
|
status = "missing_in"
|
||
|
|
elif not actual_in and not actual_out:
|
||
|
|
status = "other"
|
||
|
|
|
||
|
|
if status == "ok":
|
||
|
|
continue
|
||
|
|
rows.append(
|
||
|
|
AbnormalRow(
|
||
|
|
date=day,
|
||
|
|
rid=rid,
|
||
|
|
exp_in=exp_in,
|
||
|
|
exp_out=exp_out,
|
||
|
|
actual_in=actual_in,
|
||
|
|
actual_out=actual_out,
|
||
|
|
status=status,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
return rows
|
||
|
|
|
||
|
|
async def fetch_abnormal_rows(self, year: int, month: int) -> list[AbnormalRow]:
|
||
|
|
async with self._make_client() as client:
|
||
|
|
home = await client.get(f"{self.list_url}/")
|
||
|
|
if home.status_code in {401, 403}:
|
||
|
|
raise ValueError("CompBase NTLM login failed (401/403). Check username/password.")
|
||
|
|
home.raise_for_status()
|
||
|
|
html = home.text
|
||
|
|
|
||
|
|
data: dict[str, str] = {}
|
||
|
|
data.update(self._extract_hidden(html))
|
||
|
|
data.update(self._extract_selected_selects(html))
|
||
|
|
data.update(self._extract_checked_radios(html))
|
||
|
|
data["ddlYear"] = str(year)
|
||
|
|
data["ddlMonth"] = str(month)
|
||
|
|
if "subYear" in data:
|
||
|
|
data["subYear"] = str(year)
|
||
|
|
if "subMonth" in data:
|
||
|
|
data["subMonth"] = str(month)
|
||
|
|
# Image button click — do not send btnSet
|
||
|
|
data.pop("btnSet", None)
|
||
|
|
data.pop("Button1", None)
|
||
|
|
data["ImageCalendar.x"] = "10"
|
||
|
|
data["ImageCalendar.y"] = "10"
|
||
|
|
|
||
|
|
resp = await client.post(f"{self.list_url}/", data=data)
|
||
|
|
if resp.status_code in {401, 403}:
|
||
|
|
raise ValueError("CompBase NTLM login failed on calendar post.")
|
||
|
|
resp.raise_for_status()
|
||
|
|
rows = self._parse_table2(resp.text)
|
||
|
|
logger.info(
|
||
|
|
"CompBase calendar %04d-%02d: %d abnormal row(s)",
|
||
|
|
year,
|
||
|
|
month,
|
||
|
|
len(rows),
|
||
|
|
)
|
||
|
|
return rows
|
||
|
|
|
||
|
|
async def list_missing(self, days: int = 14) -> dict[str, Any]:
|
||
|
|
if days < 1:
|
||
|
|
raise ValueError("days must be >= 1")
|
||
|
|
today = date.today()
|
||
|
|
start = today - timedelta(days=days - 1)
|
||
|
|
months: list[tuple[int, int]] = []
|
||
|
|
cursor = date(start.year, start.month, 1)
|
||
|
|
end_month = date(today.year, today.month, 1)
|
||
|
|
while cursor <= end_month:
|
||
|
|
months.append((cursor.year, cursor.month))
|
||
|
|
if cursor.month == 12:
|
||
|
|
cursor = date(cursor.year + 1, 1, 1)
|
||
|
|
else:
|
||
|
|
cursor = date(cursor.year, cursor.month + 1, 1)
|
||
|
|
|
||
|
|
all_rows: list[AbnormalRow] = []
|
||
|
|
seen: set[tuple[date, str | None]] = set()
|
||
|
|
for year, month in months:
|
||
|
|
for row in await self.fetch_abnormal_rows(year, month):
|
||
|
|
key = (row.date, row.rid)
|
||
|
|
if key in seen:
|
||
|
|
continue
|
||
|
|
seen.add(key)
|
||
|
|
all_rows.append(row)
|
||
|
|
|
||
|
|
in_range = [r for r in all_rows if start <= r.date <= today]
|
||
|
|
fillable = [r for r in in_range if r.rid and r.status == "missing_out"]
|
||
|
|
skipped = [r for r in in_range if not (r.rid and r.status == "missing_out")]
|
||
|
|
|
||
|
|
return {
|
||
|
|
"ok": True,
|
||
|
|
"start": start.isoformat(),
|
||
|
|
"end": today.isoformat(),
|
||
|
|
"days": days,
|
||
|
|
"fillable": [r.to_dict() for r in sorted(fillable, key=lambda x: x.date)],
|
||
|
|
"skipped": [r.to_dict() for r in sorted(skipped, key=lambda x: x.date)],
|
||
|
|
"fillable_count": len(fillable),
|
||
|
|
"skipped_count": len(skipped),
|
||
|
|
}
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _span_text(html: str, element_id: str) -> str:
|
||
|
|
m = re.search(
|
||
|
|
rf'<span[^>]+id=["\']{re.escape(element_id)}["\'][^>]*>([\s\S]*?)</span>',
|
||
|
|
html,
|
||
|
|
re.I,
|
||
|
|
)
|
||
|
|
if not m:
|
||
|
|
return ""
|
||
|
|
return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", m.group(1))).strip()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _select_options(html: str, select_name: str) -> list[str]:
|
||
|
|
m = re.search(
|
||
|
|
rf'<select[^>]*name=["\']{re.escape(select_name)}["\'][^>]*>([\s\S]*?)</select>',
|
||
|
|
html,
|
||
|
|
re.I,
|
||
|
|
)
|
||
|
|
if not m:
|
||
|
|
return []
|
||
|
|
opts: list[str] = []
|
||
|
|
for om in re.finditer(r"<option[^>]*>([\s\S]*?)</option>", m.group(1), re.I):
|
||
|
|
val_m = re.search(r'value=["\']([^"\']*)["\']', om.group(0), re.I)
|
||
|
|
text = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", om.group(1))).strip()
|
||
|
|
opts.append(val_m.group(1) if val_m else text)
|
||
|
|
return opts
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _input_value(html: str, name: str) -> str:
|
||
|
|
m = re.search(
|
||
|
|
rf'<input[^>]+name=["\']{re.escape(name)}["\'][^>]*>',
|
||
|
|
html,
|
||
|
|
re.I,
|
||
|
|
)
|
||
|
|
if not m:
|
||
|
|
return ""
|
||
|
|
val = re.search(r'value=["\']([^"\']*)["\']', m.group(0), re.I)
|
||
|
|
return val.group(1) if val else ""
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _normalize_clock(value: str) -> str:
|
||
|
|
"""Normalize '5:53:00 PM' / '5:53 PM' → '5:53PM' for comparison."""
|
||
|
|
text = re.sub(r"\s+", " ", (value or "").strip()).upper()
|
||
|
|
# drop seconds if present: H:MM:SS AM/PM → H:MM AM/PM
|
||
|
|
text = re.sub(
|
||
|
|
r"\b(\d{1,2}:\d{2}):\d{2}\s*(AM|PM)\b",
|
||
|
|
r"\1 \2",
|
||
|
|
text,
|
||
|
|
)
|
||
|
|
return re.sub(r"[^0-9APM:]", "", text)
|
||
|
|
|
||
|
|
def _choose_out_time(
|
||
|
|
self,
|
||
|
|
options: list[str],
|
||
|
|
exp_out: str,
|
||
|
|
*,
|
||
|
|
out_time_mode: str,
|
||
|
|
) -> str:
|
||
|
|
usable = [
|
||
|
|
o
|
||
|
|
for o in options
|
||
|
|
if o and o not in {"------", "自行輸入"} and "自行" not in o
|
||
|
|
]
|
||
|
|
if not usable:
|
||
|
|
raise ValueError(f"No usable out-time options: {options}")
|
||
|
|
|
||
|
|
if out_time_mode == "latest_option":
|
||
|
|
return usable[-1]
|
||
|
|
|
||
|
|
exp_key = self._normalize_clock(exp_out)
|
||
|
|
for opt in usable:
|
||
|
|
if self._normalize_clock(opt) == exp_key and exp_key:
|
||
|
|
return opt
|
||
|
|
return usable[0]
|
||
|
|
|
||
|
|
async def fill_rid(
|
||
|
|
self,
|
||
|
|
rid: str,
|
||
|
|
*,
|
||
|
|
out_time_mode: str = "expected",
|
||
|
|
dry_run: bool = False,
|
||
|
|
row_date: date | None = None,
|
||
|
|
) -> dict[str, Any]:
|
||
|
|
rid = str(rid).strip()
|
||
|
|
if not rid:
|
||
|
|
raise ValueError("RID is required")
|
||
|
|
|
||
|
|
async with self._make_client() as client:
|
||
|
|
url = f"{self.fill_url}/?RID={rid}"
|
||
|
|
page = await client.get(url)
|
||
|
|
if page.status_code in {401, 403}:
|
||
|
|
raise ValueError("CompBase fill page NTLM login failed.")
|
||
|
|
page.raise_for_status()
|
||
|
|
html = page.text
|
||
|
|
|
||
|
|
exp_out = self._span_text(html, "ExpOut")
|
||
|
|
in_time = self._input_value(html, "InTime") or self._span_text(html, "ExpIn")
|
||
|
|
options = self._select_options(html, "preSetOutTimeList")
|
||
|
|
chosen = self._choose_out_time(options, exp_out, out_time_mode=out_time_mode)
|
||
|
|
label_date = self._span_text(html, "Label3") or (row_date.isoformat() if row_date else "")
|
||
|
|
|
||
|
|
result: dict[str, Any] = {
|
||
|
|
"rid": rid,
|
||
|
|
"date": label_date,
|
||
|
|
"exp_out": exp_out,
|
||
|
|
"in_time": in_time,
|
||
|
|
"chosen_out_time": chosen,
|
||
|
|
"options": options,
|
||
|
|
"dry_run": dry_run,
|
||
|
|
}
|
||
|
|
|
||
|
|
if dry_run:
|
||
|
|
result["ok"] = True
|
||
|
|
result["message"] = "dry_run"
|
||
|
|
return result
|
||
|
|
|
||
|
|
data: dict[str, str] = {}
|
||
|
|
data.update(self._extract_hidden(html))
|
||
|
|
data["RID"] = self._input_value(html, "RID") or rid
|
||
|
|
data["InTime"] = in_time
|
||
|
|
data["preSetOutTimeList"] = chosen
|
||
|
|
data["Button1"] = "送出"
|
||
|
|
|
||
|
|
resp = await client.post(url, data=data)
|
||
|
|
resp.raise_for_status()
|
||
|
|
msg = self._span_text(resp.text, "Msg")
|
||
|
|
# re-parse msg with colors if empty
|
||
|
|
if not msg:
|
||
|
|
mm = re.search(
|
||
|
|
r'id=["\']Msg["\'][^>]*>([\s\S]*?)</span>',
|
||
|
|
resp.text,
|
||
|
|
re.I,
|
||
|
|
)
|
||
|
|
if mm:
|
||
|
|
msg = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", mm.group(1))).strip()
|
||
|
|
|
||
|
|
lower = msg.lower()
|
||
|
|
failed = any(
|
||
|
|
token in msg
|
||
|
|
for token in ("失敗", "錯誤", "error", "invalid", "無法", "拒絕")
|
||
|
|
)
|
||
|
|
result["message"] = msg or "submitted"
|
||
|
|
result["ok"] = not failed
|
||
|
|
if failed:
|
||
|
|
result["error"] = msg
|
||
|
|
logger.info(
|
||
|
|
"CompBase fill RID=%s date=%s out=%s ok=%s msg=%s",
|
||
|
|
rid,
|
||
|
|
label_date,
|
||
|
|
chosen,
|
||
|
|
result["ok"],
|
||
|
|
msg,
|
||
|
|
)
|
||
|
|
return result
|
||
|
|
|
||
|
|
async def fill_recent(
|
||
|
|
self,
|
||
|
|
days: int = 14,
|
||
|
|
*,
|
||
|
|
dry_run: bool = False,
|
||
|
|
out_time_mode: str = "expected",
|
||
|
|
) -> dict[str, Any]:
|
||
|
|
scan = await self.list_missing(days)
|
||
|
|
filled: list[dict[str, Any]] = []
|
||
|
|
errors: list[dict[str, Any]] = []
|
||
|
|
skipped_rows = list(scan.get("skipped") or [])
|
||
|
|
|
||
|
|
for item in scan.get("fillable") or []:
|
||
|
|
rid = item.get("rid")
|
||
|
|
if not rid:
|
||
|
|
skipped_rows.append({**item, "reason": "no_rid"})
|
||
|
|
continue
|
||
|
|
try:
|
||
|
|
day = date.fromisoformat(item["date"]) if item.get("date") else None
|
||
|
|
result = await self.fill_rid(
|
||
|
|
str(rid),
|
||
|
|
out_time_mode=out_time_mode,
|
||
|
|
dry_run=dry_run,
|
||
|
|
row_date=day,
|
||
|
|
)
|
||
|
|
if result.get("ok"):
|
||
|
|
filled.append({**item, **result})
|
||
|
|
else:
|
||
|
|
errors.append({**item, **result})
|
||
|
|
except Exception as exc: # noqa: BLE001
|
||
|
|
logger.exception("CompBase fill failed for RID=%s", rid)
|
||
|
|
errors.append({**item, "ok": False, "error": str(exc)})
|
||
|
|
|
||
|
|
# annotate skipped without rid
|
||
|
|
for row in skipped_rows:
|
||
|
|
row.setdefault("reason", "no_rid" if not row.get("rid") else row.get("status", "skipped"))
|
||
|
|
|
||
|
|
return {
|
||
|
|
"ok": len(errors) == 0,
|
||
|
|
"dry_run": dry_run,
|
||
|
|
"days": days,
|
||
|
|
"out_time_mode": out_time_mode,
|
||
|
|
"start": scan.get("start"),
|
||
|
|
"end": scan.get("end"),
|
||
|
|
"fillable_count": scan.get("fillable_count"),
|
||
|
|
"filled": filled,
|
||
|
|
"skipped": skipped_rows,
|
||
|
|
"errors": errors,
|
||
|
|
"filled_count": len(filled),
|
||
|
|
"skipped_count": len(skipped_rows),
|
||
|
|
"error_count": len(errors),
|
||
|
|
}
|
||
|
|
|
||
|
|
async def status(self) -> dict[str, Any]:
|
||
|
|
username = (getattr(settings, "compbase_username", "") or "").strip()
|
||
|
|
source = "compbase"
|
||
|
|
if not username:
|
||
|
|
username = (getattr(settings, "pts_username", "") or "").strip()
|
||
|
|
source = "pts" if username else "none"
|
||
|
|
has_password = bool(
|
||
|
|
(getattr(settings, "compbase_password", "") or "")
|
||
|
|
or (getattr(settings, "pts_password", "") or "")
|
||
|
|
)
|
||
|
|
result: dict[str, Any] = {
|
||
|
|
"list_url": self.list_url,
|
||
|
|
"fill_url": self.fill_url,
|
||
|
|
"username": username or None,
|
||
|
|
"credential_source": source,
|
||
|
|
"has_credentials": bool(username and has_password),
|
||
|
|
"default_days": int(getattr(settings, "compbase_default_days", 14) or 14),
|
||
|
|
"out_time_mode": getattr(settings, "compbase_out_time_mode", "expected") or "expected",
|
||
|
|
"connected": False,
|
||
|
|
}
|
||
|
|
if not result["has_credentials"]:
|
||
|
|
result["error"] = "No credentials configured"
|
||
|
|
return result
|
||
|
|
try:
|
||
|
|
async with self._make_client() as client:
|
||
|
|
resp = await client.get(f"{self.list_url}/")
|
||
|
|
result["connected"] = resp.status_code == 200
|
||
|
|
result["http_status"] = resp.status_code
|
||
|
|
if resp.status_code == 200:
|
||
|
|
# pull employee name if present
|
||
|
|
m = re.search(r'id=["\']CName["\'][^>]*>([^<]+)', resp.text)
|
||
|
|
if m:
|
||
|
|
result["display_name"] = m.group(1).strip()
|
||
|
|
m = re.search(r'id=["\']EmpID["\'][^>]*>([^<]+)', resp.text)
|
||
|
|
if m:
|
||
|
|
result["emp_id"] = m.group(1).strip()
|
||
|
|
else:
|
||
|
|
result["error"] = f"HTTP {resp.status_code}"
|
||
|
|
except Exception as exc: # noqa: BLE001
|
||
|
|
result["error"] = str(exc)
|
||
|
|
return result
|