"""Pure, explicit user scope for read-only relationship preflight tasks.""" from __future__ import annotations from dataclasses import dataclass import re from urllib.parse import urlsplit @dataclass(frozen=True) class PreflightPolicy: active: bool = False excluded_note_ids: tuple[str, ...] = () max_notes: int | None = None _NOTE_PATH = re.compile(r"/(?:explore|discovery/item)/([a-f0-9]{24})/?", re.I) _ID = re.compile(r"(?\"'`]+(?:\?[^\s<>\"'`]+)?", re.I) _LITERAL = r"(?:[a-fA-F0-9]{24}|save_source|save_learning|https://(?:www\.)?xiaohongshu\.com/(?:explore|discovery/item)/[a-fA-F0-9]{24}[^\s，,。；;！？`\"'“”‘’「」『』]*)" def note_id_from_url(actual_url: str) -> str | None: """Read identity from an actual XHS note URL; never create a destination.""" try: parts = urlsplit(actual_url) if (parts.scheme != "https" or parts.hostname not in {"www.xiaohongshu.com", "xiaohongshu.com"} or parts.username or parts.password or parts.port not in {None, 443}): return None match = _NOTE_PATH.fullmatch(parts.path) return match.group(1).lower() if match else None except (ValueError, TypeError, AttributeError): return None def instruction_text(goal: str) -> str: """Keep formatted constants, not quoted requests, examples or documents.""" text = re.sub(r"```[\s\S]*?```", "", goal) text = "\n".join('' if line.lstrip().startswith('>') else line for line in text.splitlines()) for left, right in (("`", "`"), ('"', '"'), ("'", "'"), ("“", "”"), ("‘", "’"), ("「", "」"), ("『", "』")): text = re.sub(re.escape(left) + "(" + _LITERAL + ")" + re.escape(right), r"\1", text) return re.sub(r'''“[^”]*”|‘[^’]*’|「[^」]*」|『[^』]*』|"[^"\n]*"|'[^'\n]*'|`[^`\n]*`''', "", text) def _clauses(goal: str): # URLs are target data, including their query punctuation and words. Keep # them atomic while splitting sentences, and exclude them from directives. urls = [] def protect(match): value = match.group() suffix = re.search(r'[，,。；;！？!?]+$', value) ending = suffix.group() if suffix else '' urls.append(value[:-len(ending)] if ending else value) return '\ue000' + str(len(urls)-1) + '\ue001' + ending text = _URL.sub(protect, instruction_text(goal)) boundaries = r"[，,。；;！？!?\n]|但是|不过|而是|但" for clause in re.split('(' + boundaries + ')', text): if re.fullmatch(boundaries, clause): yield '', '', clause continue directive = re.sub(r'\ue000[0-9]+\ue001', '', clause).strip() value = re.sub(r'\ue000([0-9]+)\ue001', lambda match: urls[int(match.group(1))], clause).strip() if value and not re.search(r"例如|比如|示例|参考案例|只是.{0,8}(?:引用|日志|描述)|(?:日志|文档|截图|网页).{0,8}(?:写着|显示|提示|内容)|(?:解释|讨论|介绍).{0,8}(?:预检|不保存)", directive): yield value, directive, '' elif value: yield '', '', 'ignored' _ONLY = re.compile(r"只读预检|(?:只|仅)(?:做|进行)?(?:(?:只读)?(?:(?:二级)?回复关系|关系|回复)?|(?:(?:二级)?回复关系|关系|回复)的?只读)预检") _NEGATIVE_BEFORE = re.compile(r"(?:不是|并非|不再|不做|不进行|不要|不用|无需|不必|不需要|取消|停止|放弃|先不|暂不|别)[^，,。；;]{0,10}$") _NEGATIVE_AFTER = re.compile(r"\s*(?:先|暂时)?(?:取消|不用|无需|不要|不必|停止|不做)") _SAVE = re.compile(r"保存(?:来源|原话|资料|笔记|评论|回复|证据)?|save_source") _NO_SAVE_BEFORE = re.compile(r"(?:不|不要|不用|无需|不必|不需要|禁止|不允许|不得|别|不能)(?:再|实际|新增|任何|进行|调用|使用|去|做|将|把|本轮|本次|这些|所有|来源|原话|资料|笔记|评论|回复|证据|\s){0,12}$") _POSITIVE_SAVE = re.compile(r"(?:改为|改成|开始|现在|然后|最后|后面|稍后|随后|之后|再|后|需要|可以|允许|要|请|执行|正常|直接|仍要|继续).{0,8}$") _EXCLUDE = re.compile(r"排除|跳过|(?:不|不要|不用|无需|不必|不需要|不允许|不能|别|禁止|不得)(?:再)?(?:打开|访问|阅读|进入|读|看|预检)") _INCLUDE = re.compile(r"(?:不|不要|不用|无需|不必|不需要|不再|取消|停止|别)(?:再)?(?:排除|跳过)|(? int | None: if value.isascii() and value.isdigit(): return int(value) digits = {"一": 1, "二": 2, "两": 2, "三": 3, "四": 4, "五": 5, "六": 6, "七": 7, "八": 8, "九": 9} total, current = 0, 0 for char in value: if char in digits: if current: return None current = digits[char] elif char in {"十", "百"}: total += (current or 1) * {"十": 10, "百": 100}[char] current = 0 else: return None return total + current or None def _note_ids(clause: str, directive: str) -> list[str]: found, spans = [], [] for match in _URL.finditer(clause): spans.append(match.span()) note_id = note_id_from_url(match.group()) if note_id and note_id not in found: found.append(note_id) # Account/profile IDs and IDs embedded in unsupported URLs are not notes. account_clause = bool(re.search(r"用户|账号|作者|发言人|主页|profile", directive, re.I)) and not re.search(r"笔记|note|旧篇|旧样本", directive, re.I) if not account_clause: for match in _ID.finditer(clause): if not any(start <= match.start() < end for start, end in spans): note_id = match.group(1).lower() if note_id not in found: found.append(note_id) return found def compile_preflight_policy(goal: str, max_notes: int | None = None) -> PreflightPolicy: """Only explicit preflight + no source persistence activates this policy. ``max_notes`` is accepted for the caller's existing brief contract. It never creates or replaces a limit: the actual human instruction is authoritative. No old task, source, model output or prior visit is consulted or modified. """ only, no_save, cap = False, False, None excluded: list[str] = [] exclusion_continues = list_mode = False line_has_clause = False for raw_clause, clause, boundary in _clauses(goal): if boundary: if boundary == '\n': if not list_mode or not line_has_clause: exclusion_continues = list_mode = False line_has_clause = False elif boundary not in {'，', ','}: exclusion_continues = list_mode = False continue line_has_clause = True for match in _ONLY.finditer(clause): if re.search(r"(?:每|各)(?:篇|条|组|位)(?:先)?(?:做|进行)?(?:一轮|一次)?\s*$", clause[:match.start()]): continue only = not (_NEGATIVE_BEFORE.search(clause[:match.start()]) or _NEGATIVE_AFTER.match(clause[match.end():])) if re.search(r"(?:取消|停止|放弃).{0,8}(?:只读)?(?:关系)?预检|(?:改为|改成).{0,8}(?:正式|正常|完整)取证", clause): only = False for match in _SAVE.finditer(clause): before, after = clause[:match.start()], clause[match.end():] if match.group() == "保存" and re.match(r"\s*(?:任何|本次|这些)?(?:学习|经验|知识|日志|文件|截图|草稿|配置|模型)", after): continue if _NO_SAVE_BEFORE.search(before): no_save = not bool(_NEGATIVE_AFTER.match(after)) elif _POSITIVE_SAVE.search(before) or re.match(r"\s*(?:要|可以|允许|需要|照常|正常)做?", after): no_save = False if re.search(r"(?:不|无需|不用|不必|不要|别)(?:再)?(?:限制|限定)(?:笔记)?篇数|篇数(?:不限|不限制|不限定)", clause): cap = None for match in _CAP.finditer(clause): if (_NEGATIVE_BEFORE.search(clause[:match.start()]) or re.match(r"\s*(?:左右|以上|以下|到|至|或|[-~～/])", clause[match.end():])): continue cap = _number(match.group(1)) if re.search(r"(?:不|不用|无需|取消).{0,5}排除(?:任何|所有|全部)笔记|排除(?:项|列表)(?:清空|取消)", clause): excluded.clear() exclusion_continues = list_mode = False ids = _note_ids(raw_clause, clause) include = bool(_INCLUDE.search(clause)) exclude = bool(_EXCLUDE.search(clause)) and not include if include: excluded = [note_id for note_id in excluded if note_id not in ids] exclusion_continues = list_mode = False elif exclude: if re.search(r"排除(?:项|列表)?(?:改为|改成|只有|仅有)", clause): excluded.clear() excluded.extend(note_id for note_id in ids if note_id not in excluded) exclusion_continues = True list_mode = bool(_EXCLUSION_LIST.search(clause)) elif ids and exclusion_continues and not _NEW_INSTRUCTION.search(clause): excluded.extend(note_id for note_id in ids if note_id not in excluded) elif not ids or _NEW_INSTRUCTION.search(clause): exclusion_continues = list_mode = False if not only or not no_save: return PreflightPolicy() return PreflightPolicy(active=True, excluded_note_ids=tuple(excluded), max_notes=cap)