"""Khung dùng chung cho các chiến dịch gửi hàng loạt (.thongbao, .lead). Chỉ chứa phần hai module trùng nhau từng chữ: khóa chống chạy song song, ghi state an toàn, và các helper chuẩn hóa. Vòng đời chiến dịch của mỗi lệnh vẫn nằm trong module của nó vì chúng khác nhau thật. Mọi hàm nhận đường dẫn qua callable để test vẫn patch được STATE_FILE. """ import json import os import re import tempfile import threading from datetime import datetime from zlapi.models import Message from sales_utils import split_text try: import fcntl except ImportError: # pragma: no cover - hosting chạy Unix fcntl = None def now_iso(): return datetime.now().astimezone().isoformat(timespec="seconds") def normalize_id(value): if value is None or isinstance(value, bool): return None normalized = str(value).strip() return normalized or None def iter_ids(value): if value is None: return [] if isinstance(value, (str, int)): return [value] try: return list(value) except TypeError: return [value] def is_admin(author_id, config_admin, client=None): allowed = {normalize_id(value) for value in iter_ids(config_admin)} if client is not None: allowed.update(normalize_id(value) for value in iter_ids(getattr(client, "ADMIN", None))) allowed.update(normalize_id(value) for value in iter_ids(getattr(client, "ADM", None))) allowed.discard(None) return normalize_id(author_id) in allowed def value_of(source, key, default=None): if source is None: return default if isinstance(source, dict): return source.get(key, default) return getattr(source, key, default) def error_text(error): text = str(error or error.__class__.__name__).strip() return (text or error.__class__.__name__)[:500] def validate_delay(raw_value, minimum, maximum): if isinstance(raw_value, bool): raise ValueError("Thời gian nghỉ phải là số nguyên") text = str(raw_value or "").strip() if not re.fullmatch(r"\d+", text): raise ValueError("Thời gian nghỉ phải là số nguyên") delay = int(text) if not minimum <= delay <= maximum: raise ValueError(f"Thời gian nghỉ phải từ {minimum} đến {maximum} giây") return delay def flag_enabled(value): """Cờ bật/tắt từ Zalo có thể là bool, số, hoặc chuỗi "0"/"1"/"true".""" if isinstance(value, bool): return value if isinstance(value, (int, float)): return value != 0 return str(value or "").strip().casefold() in {"1", "true", "yes", "on"} def write_json_atomic(path, data, prefix): """Ghi qua file tạm + fsync + os.replace: state không bao giờ ghi dở.""" directory = os.path.dirname(path) or "." os.makedirs(directory, exist_ok=True) temp_file = tempfile.NamedTemporaryFile( mode="w", encoding="utf-8", dir=directory, prefix=prefix, suffix=".tmp", delete=False ) try: json.dump(data, temp_file, ensure_ascii=False, indent=2) temp_file.flush() os.fsync(temp_file.fileno()) temp_file.close() os.replace(temp_file.name, path) except BaseException: temp_file.close() try: os.remove(temp_file.name) except OSError: pass raise class CrossProcessRLock: """RLock theo thread + flock theo process, khóa quanh file state. Cần flock vì bot có thể bị hosting khởi động lại chồng tiến trình; hai tiến trình cùng đọc-sửa-ghi state là gửi trùng cho khách. """ def __init__(self, lock_path): self._lock_path = lock_path # callable: STATE_FILE đổi được khi test self._thread_lock = threading.RLock() self._local = threading.local() def __enter__(self): self.acquire() return self def __exit__(self, exc_type, exc_value, traceback): self.release() def acquire(self): self._thread_lock.acquire() depth = getattr(self._local, "depth", 0) try: if depth == 0 and fcntl is not None: lock_path = self._lock_path() os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True) lock_handle = open(lock_path, "a+", encoding="utf-8") try: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX) except BaseException: lock_handle.close() raise self._local.lock_handle = lock_handle self._local.depth = depth + 1 return True except BaseException: self._thread_lock.release() raise def release(self): depth = getattr(self._local, "depth", 0) if depth <= 0: raise RuntimeError("State lock is not owned by this thread") next_depth = depth - 1 try: if next_depth == 0: lock_handle = getattr(self._local, "lock_handle", None) if lock_handle is not None: try: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN) finally: lock_handle.close() del self._local.lock_handle del self._local.depth else: self._local.depth = next_depth finally: self._thread_lock.release() class RunLock: """Chỉ cho một chiến dịch chạy tại một thời điểm, kể cả khác tiến trình. `active` và `handles` để lộ ra ngoài vì module gọi và test đọc trực tiếp. """ def __init__(self, lock_path, state_lock, already_running_message, busy_message): self._lock_path = lock_path # callable self._state_lock = state_lock self._already_running_message = already_running_message self._busy_message = busy_message self.active = set() self.handles = {} def claim(self, campaign_id): with self._state_lock: if campaign_id in self.active: raise ValueError(self._already_running_message) if self.active: raise ValueError(self._busy_message) lock_handle = None if fcntl is not None: lock_path = self._lock_path() os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True) lock_handle = open(lock_path, "a+", encoding="utf-8") try: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as error: lock_handle.close() raise ValueError(self._busy_message) from error except OSError: lock_handle.close() raise self.active.add(campaign_id) if lock_handle is not None: self.handles[campaign_id] = lock_handle def release(self, campaign_id): with self._state_lock: self.active.discard(campaign_id) lock_handle = self.handles.pop(campaign_id, None) if lock_handle is not None: try: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN) finally: lock_handle.close() def is_held(self): if self.active: return True if fcntl is None: return False lock_path = self._lock_path() try: os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True) with open(lock_path, "a+", encoding="utf-8") as lock_handle: try: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: return True fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN) return False except OSError: # Fail closed: không rõ ai giữ khóa thì coi như đang chạy. return True def reply(client, text, message_object, thread_id, thread_type, limit=2800, ttl=120000): for chunk in split_text(str(text or ""), limit): client.replyMessage(Message(text=chunk), message_object, thread_id, thread_type, ttl=ttl) if __name__ == "__main__": # self-check: khóa tái nhập, ghi atomic, và chống chạy chồng chiến dịch import tempfile as _tempfile assert normalize_id(" 42 ") == "42" and normalize_id(True) is None and normalize_id("") is None assert iter_ids(None) == [] and iter_ids("a") == ["a"] and iter_ids(["a", "b"]) == ["a", "b"] assert is_admin("1", ["1"]) and not is_admin("2", ["1"]) assert flag_enabled(1) and flag_enabled("1") and flag_enabled(True) and flag_enabled("on") for falsy in (0, "0", "", None, False, "off", "false"): assert not flag_enabled(falsy), f"{falsy!r} phải là tắt" assert validate_delay("5", 2, 300) == 5 for bad in ("0", "abc", "999", True): try: validate_delay(bad, 2, 300) raise AssertionError(f"phải chặn {bad!r}") except ValueError: pass directory = _tempfile.mkdtemp() state_path = os.path.join(directory, "state.json") write_json_atomic(state_path, {"ok": "có dấu"}, "selfcheck_") assert json.load(open(state_path, encoding="utf-8")) == {"ok": "có dấu"} assert not [n for n in os.listdir(directory) if n.endswith(".tmp")], "còn file tạm" state_lock = CrossProcessRLock(lambda: f"{state_path}.lock") with state_lock: with state_lock: # tái nhập được pass run_lock = RunLock(lambda: f"{state_path}.run.lock", state_lock, "đang chạy", "bận") run_lock.claim("c1") assert run_lock.is_held() for campaign_id, expected in (("c1", "đang chạy"), ("c2", "bận")): try: run_lock.claim(campaign_id) raise AssertionError("phải chặn claim thứ hai") except ValueError as error: assert str(error) == expected run_lock.release("c1") assert not run_lock.active and not run_lock.handles print("campaign self-check OK")