forked from erp-dev/erp
92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
import re
|
|
|
|
from mission.models import MissionPayloadProcessorEnum
|
|
|
|
|
|
STRUCTURED_DESCRIPTION_IMAGE_PREFIXES = ("款式图:", "款式图：")
|
|
|
|
|
|
def _extract_labeled_image_url(lines: list[str]) -> tuple[list[str], str]:
|
|
kept_lines = []
|
|
image_url = ""
|
|
for line in lines:
|
|
stripped_line = line.strip()
|
|
matched_prefix = None
|
|
for prefix in STRUCTURED_DESCRIPTION_IMAGE_PREFIXES:
|
|
if stripped_line.startswith(prefix):
|
|
matched_prefix = prefix
|
|
break
|
|
if matched_prefix is None:
|
|
kept_lines.append(line)
|
|
continue
|
|
|
|
candidate = stripped_line.removeprefix(matched_prefix).strip()
|
|
if candidate:
|
|
image_url = candidate
|
|
return kept_lines, image_url
|
|
|
|
|
|
def _split_structured_description(description: str) -> tuple[str, str, str, str]:
|
|
lines = [line.strip() for line in (description or "").splitlines()]
|
|
if not lines:
|
|
return "", "", "", ""
|
|
|
|
body_lines = lines[1:] if len(lines) > 1 else []
|
|
while body_lines and not body_lines[0]:
|
|
body_lines.pop(0)
|
|
|
|
body_lines, image_url = _extract_labeled_image_url(body_lines)
|
|
|
|
url = ""
|
|
if body_lines:
|
|
last_line = body_lines[-1]
|
|
match = re.search(r"https?://\S+", last_line)
|
|
if match is not None:
|
|
url = match.group(0)
|
|
body_lines = body_lines[:-1]
|
|
|
|
while body_lines and not body_lines[-1]:
|
|
body_lines.pop()
|
|
|
|
split_index = next((index for index, line in enumerate(body_lines) if not line), None)
|
|
if split_index is None:
|
|
title_lines = body_lines
|
|
description_lines = []
|
|
else:
|
|
title_lines = body_lines[:split_index]
|
|
description_lines = body_lines[split_index + 1 :]
|
|
while description_lines and not description_lines[0]:
|
|
description_lines.pop(0)
|
|
|
|
title = "\n".join(title_lines).strip()
|
|
parsed_description = "\n".join(description_lines).strip()
|
|
return title, parsed_description, url, image_url
|
|
|
|
|
|
def enhance_structured_description_v1(*, mission, payload: dict, event_key: str) -> dict:
|
|
title, parsed_description, url, image_url = _split_structured_description(mission.description)
|
|
return {
|
|
**payload,
|
|
"parsed_description_title": title,
|
|
"parsed_description_body": parsed_description,
|
|
"parsed_description_url": url,
|
|
"parsed_description_image_url": image_url,
|
|
}
|
|
|
|
|
|
PAYLOAD_PROCESSORS = {
|
|
MissionPayloadProcessorEnum.STRUCTURED_DESCRIPTION_V1: enhance_structured_description_v1,
|
|
}
|
|
|
|
|
|
def apply_mission_payload_processor(*, mission, payload: dict, event_key: str) -> dict:
|
|
processor_key = getattr(mission.category, "payload_processor", "") or ""
|
|
if not processor_key:
|
|
return payload
|
|
|
|
processor = PAYLOAD_PROCESSORS.get(processor_key)
|
|
if processor is None:
|
|
return payload
|
|
|
|
enhanced_payload = processor(mission=mission, payload=payload, event_key=event_key)
|
|
return enhanced_payload if isinstance(enhanced_payload, dict) else payload |