182 lines
10 KiB
Python
182 lines
10 KiB
Python
"""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"(?<![a-f0-9])([a-f0-9]{24})(?![a-f0-9])", re.I)
|
||
_URL = re.compile(r"https?://[^\s，,。；;！？!?<>\"'`]+(?:\?[^\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"(?:不|不要|不用|无需|不必|不需要|不再|取消|停止|别)(?:再)?(?:排除|跳过)|(?<!不)(?:允许|可以|改为|改成|需要)(?:再)?(?:打开|访问|阅读|进入|预检|包含)")
|
||
_NUM = r"(?:[1-9][0-9]*|[一二两三四五六七八九十百]+)"
|
||
_CAP = re.compile(r"(?:最多|至多|不超过|上限(?:是|为)?)(?:只)?(?:浏览|阅读|访问|检查|预检|查看|看|打开|挑选|找|寻找)?\s*(?:新的|新|不同的|不同|候选)?(?:笔记)?\s*(" + _NUM + r")\s*篇(?:新的|新|不同的|不同)?(?:候选)?(?:笔记)?")
|
||
_EXCLUSION_LIST = re.compile(r"(?:排除|跳过).{0,12}(?:以下|如下|下列|列表|清单).{0,8}[:：]")
|
||
_NEW_INSTRUCTION = re.compile(r"保存|预检|打开|访问|阅读|检查|然后|最后|目标|入口|链接|当前页面|实际页面|先沿|只|仅|最多|至多|上限")
|
||
|
||
|
||
def _number(value: str) -> 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)
|