Spaces:
Sleeping
Sleeping
| """ | |
| ์ ์ฑ yaml ๋ค์ค ๋ก๋ + ์ ์ฑ ๋ณ ํ๊ฐ + ํฉ์ฑ. | |
| ยง50 ์ฐ์ ์์: opt_out > no_consent(6mo ์์ธ) > night > ALLOWED | |
| ยง15ยทยง16 ์ฐ์ ์์: purpose_violation > retention_expired > ALLOWED | |
| ํฉ์ฑ: ๋ชจ๋ ALLOWED์ฌ์ผ ALLOWED, ํ๋๋ผ๋ BLOCK์ด๋ฉด ์ ์ฑ ๋ณ ์ฌ์ ๋์ (๊ฐ์ฅ ๊ฐํ BLOCK ์ฒซ ๋ฒ์งธ). | |
| """ | |
| from __future__ import annotations | |
| import pathlib | |
| import yaml | |
| _BLOCK_PRIORITY = { | |
| "BLOCKED_OPT_OUT": 0, | |
| "BLOCKED_PURPOSE": 1, | |
| "BLOCKED_RETENTION": 2, | |
| "BLOCKED_NO_CONSENT": 3, | |
| "BLOCKED_NIGHT": 4, | |
| } | |
| def load_policies() -> dict[str, dict]: | |
| """`policies/*.yaml` ๋ชจ๋ ๋ก๋ โ {name: policy_dict}.""" | |
| base = pathlib.Path(__file__).parent / "policies" | |
| out: dict[str, dict] = {} | |
| for path in sorted(base.glob("*.yaml")): | |
| with open(path, encoding="utf-8") as f: | |
| data = yaml.safe_load(f) | |
| out[data["name"]] = data | |
| return out | |
| def evaluate_infomatics(r: dict, *, send_hour: int, policy: dict) -> dict: | |
| """ยง50: opt_out > no_consent > night > ALLOWED.""" | |
| citation = policy["citation"] | |
| rid = r.get("id") | |
| if r["opt_out"]: | |
| return {"status": "BLOCKED_OPT_OUT", "citation": citation, "id": rid, "policy": "infomatics"} | |
| if not r["marketing_consent"]: | |
| last = r.get("last_same_product_days") | |
| if last is None or last > policy["same_product_exception_days"]: | |
| return {"status": "BLOCKED_NO_CONSENT", "citation": citation, "id": rid, "policy": "infomatics"} | |
| return {"status": "ALLOWED_EXCEPTION", "citation": citation, "id": rid, "policy": "infomatics"} | |
| start = policy["night_window"]["start_hour"] | |
| end = policy["night_window"]["end_hour"] | |
| is_night = send_hour >= start or send_hour < end | |
| if is_night and not r["night_consent"]: | |
| return {"status": "BLOCKED_NIGHT", "citation": citation, "id": rid, "policy": "infomatics"} | |
| return {"status": "ALLOWED", "citation": citation, "id": rid, "policy": "infomatics"} | |
| def evaluate_pipa(r: dict, policy: dict) -> dict: | |
| """ยง15ยทยง16: purpose_violation > retention_expired > ALLOWED.""" | |
| citation = policy["citation"] | |
| rid = r.get("id") | |
| purpose = r.get("collected_purpose", "marketing") | |
| if purpose not in policy["allowed_purposes"]: | |
| return {"status": "BLOCKED_PURPOSE", "citation": citation, "id": rid, "policy": "pipa"} | |
| days = r.get("collected_days_ago", 0) | |
| if days > policy["retention_max_days"]: | |
| return {"status": "BLOCKED_RETENTION", "citation": citation, "id": rid, "policy": "pipa"} | |
| return {"status": "ALLOWED", "citation": citation, "id": rid, "policy": "pipa"} | |
| def evaluate_recipient(r: dict, *, send_hour: int, policies: dict[str, dict]) -> dict: | |
| """๋ค์ค ์ ์ฑ ํฉ์ฑ. ALLOWED iff ๋ชจ๋ ์ ์ฑ ALLOWED. ์๋๋ฉด ๊ฐ์ฅ ๊ฐํ BLOCK ์ฑํ.""" | |
| results = [] | |
| if "infomatics" in policies: | |
| results.append(evaluate_infomatics(r, send_hour=send_hour, policy=policies["infomatics"])) | |
| if "pipa" in policies: | |
| results.append(evaluate_pipa(r, policies["pipa"])) | |
| blocks = [x for x in results if x["status"].startswith("BLOCKED_")] | |
| if not blocks: | |
| # ALLOWED ๋๋ ALLOWED_EXCEPTION ์ค ๊ฐ์ฅ ๊ฐํ ๋ผ๋ฒจ ์ฑํ(EXCEPTION ์ฐ์ ํ๊ธฐ) | |
| if any(x["status"] == "ALLOWED_EXCEPTION" for x in results): | |
| return next(x for x in results if x["status"] == "ALLOWED_EXCEPTION") | |
| return results[0] # ALLOWED | |
| blocks.sort(key=lambda x: _BLOCK_PRIORITY.get(x["status"], 999)) | |
| primary = blocks[0] | |
| primary["all_blocks"] = blocks # ๋๋ฒ๊น ยทreport ์ฉ | |
| return primary | |
| def evaluate_recipient_legacy(r: dict, send_hour: int, policy: dict) -> dict: | |
| """03 ํธํ wrapper โ ๋จ์ผ infomatics ์ ์ฑ ์๊ทธ๋์ฒ.""" | |
| return evaluate_infomatics(r, send_hour=send_hour, policy=policy) | |