| """开庭时间提取 Mixin""" |
|
|
| import logging |
| from datetime import datetime |
| from typing import Any, cast |
|
|
| logger = logging.getLogger("apps.document_recognition") |
|
|
| |
| DATETIME_PATTERNS = [ |
| (r"(\d{4})年(\d{1,2})月(\d{1,2})日\s+(上午|下午)\s*(\d{1,2})[时点](\d{1,2})分?", True), |
| (r"(\d{4})年(\d{1,2})月(\d{1,2})日(上午|下午)(\d{1,2})[时点](\d{1,2})分?", True), |
| (r"(\d{4})年(\d{1,2})月(\d{1,2})日\s+(\d{1,2})[时点](\d{1,2})分?", False), |
| (r"(\d{4})年(\d{1,2})月(\d{1,2})日(\d{1,2})[时点](\d{1,2})分?", False), |
| (r"(\d{4})年(\d{1,2})月(\d{1,2})日\s*(\d{1,2}):(\d{1,2})", False), |
| (r"(\d{4})-(\d{1,2})-(\d{1,2})\s+(\d{1,2}):(\d{1,2})", False), |
| (r"(\d{4})/(\d{1,2})/(\d{1,2})\s+(\d{1,2}):(\d{1,2})", False), |
| ] |
|
|
| HEARING_HIGH_WEIGHT_KEYWORDS = ["开庭", "庭审", "定于", "传唤"] |
| HEARING_MEDIUM_WEIGHT_KEYWORDS = [ |
| "审理", |
| "出庭", |
| "到庭", |
| "准时到", |
| "按时到", |
| "应诉", |
| "参加诉讼", |
| "审判庭", |
| "法庭", |
| "第一审判庭", |
| "第二审判庭", |
| "第三审判庭", |
| "应到时间", |
| "到庭时间", |
| "开庭时间", |
| ] |
| HEARING_LOW_WEIGHT_KEYWORDS = ["本院", "通知", "届时", "如期", "依法"] |
| HEARING_CONTEXT_KEYWORDS = HEARING_HIGH_WEIGHT_KEYWORDS + HEARING_MEDIUM_WEIGHT_KEYWORDS + HEARING_LOW_WEIGHT_KEYWORDS |
|
|
|
|
| class DatetimeExtractionMixin: |
| """开庭时间提取 Mixin""" |
|
|
| import re as _re |
|
|
| def _parse_datetime_groups(self, groups: tuple[Any, ...], has_am_pm: bool, matched_text: str) -> datetime | None: |
| """解析正则分组为 datetime""" |
| if has_am_pm and len(groups) == 6: |
| year, month, day = int(groups[0]), int(groups[1]), int(groups[2]) |
| am_pm, hour, minute = groups[3], int(groups[4]), int(groups[5]) |
| if am_pm not in ("上午", "下午"): |
| logger.debug(f"无效的上午/下午标识: {am_pm}, 跳过匹配: {matched_text}") |
| return None |
| if am_pm == "下午" and hour < 12: |
| hour += 12 |
| elif am_pm == "上午" and hour == 12: |
| hour = 0 |
| elif len(groups) == 5: |
| year, month, day = int(groups[0]), int(groups[1]), int(groups[2]) |
| hour, minute = int(groups[3]), int(groups[4]) |
| else: |
| logger.debug(f"未知的分组数量: {len(groups)}, 跳过匹配: {matched_text}") |
| return None |
| if not (1 <= month <= 12 and 1 <= day <= 31 and 0 <= hour <= 23 and 0 <= minute <= 59): |
| logger.debug(f"日期时间无效: {matched_text}") |
| return None |
| if not (2020 <= year <= 2030): |
| logger.debug(f"年份不合理: {year}, 跳过匹配: {matched_text}") |
| return None |
| return datetime(year, month, day, hour, minute) |
|
|
| def _extract_datetime_by_regex(self, text: str) -> list[tuple[datetime, str, float]]: |
| """使用正则表达式从文本中提取日期时间""" |
| import re |
|
|
| results: list[tuple[datetime, str, float]] = [] |
| for pattern, has_am_pm in DATETIME_PATTERNS: |
| for match in re.finditer(pattern, text): |
| try: |
| matched_text = match.group(0) |
| dt = self._parse_datetime_groups(match.groups(), has_am_pm, matched_text) |
| if dt is None: |
| continue |
| context_score = self._calculate_context_score(text, match.start()) |
| if not any(abs((dt - existing[0]).total_seconds()) < 60 for existing in results): |
| results.append((dt, matched_text, context_score)) |
| logger.debug(f"正则提取到时间: {dt}, 原文: {matched_text}, 上下文得分: {context_score}") |
| except (ValueError, IndexError) as e: |
| logger.debug(f"解析日期时间失败: {match.group(0)}, 错误: {e}") |
| continue |
| return results |
|
|
| def _calculate_context_score(self, text: str, match_position: int) -> int: |
| """计算匹配位置的上下文得分（0-100）""" |
| context_start = max(0, match_position - 100) |
| context_end = min(len(text), match_position + 50) |
| context = text[context_start:context_end] |
| score = 0 |
| for keyword in HEARING_HIGH_WEIGHT_KEYWORDS: |
| if keyword in context: |
| score += 25 |
| for keyword in HEARING_MEDIUM_WEIGHT_KEYWORDS: |
| if keyword in context: |
| score += 15 |
| for keyword in HEARING_LOW_WEIGHT_KEYWORDS: |
| if keyword in context: |
| score += 8 |
| return min(score, 100) |
|
|
| def _score_days_diff(self, days_diff: int, score: int, reasons: list[str]) -> tuple[int, list[str]]: |
| """评估天数差对分数的影响""" |
| if days_diff < -7: |
| reasons.append(f"时间已过去{abs(days_diff)}天") |
| score -= 30 |
| elif days_diff < 0: |
| reasons.append(f"时间已过去{abs(days_diff)}天") |
| score -= 10 |
| elif days_diff > 365 * 2: |
| reasons.append(f"时间在{days_diff}天后，超过2年") |
| score -= 40 |
| elif days_diff > 365: |
| reasons.append(f"时间在{days_diff}天后") |
| score -= 15 |
| elif days_diff > 180: |
| reasons.append(f"时间在{days_diff}天后") |
| score -= 5 |
| else: |
| reasons.append(f"时间在{days_diff}天后") |
| score += 20 |
| return score, reasons |
|
|
| def _validate_hearing_datetime(self, dt: datetime) -> tuple[bool, int, str]: |
| """校验开庭时间的合理性""" |
| now = datetime.now() |
| score = 50 |
| reasons: list[str] = [] |
| days_diff = (dt.date() - now.date()).days |
| score, reasons = self._score_days_diff(days_diff, score, reasons) |
| hour = dt.hour |
| if 8 <= hour <= 18: |
| score += 15 |
| reasons.append("工作时间内") |
| elif 7 <= hour < 8 or 18 < hour <= 20: |
| score += 5 |
| reasons.append("边缘工作时间") |
| else: |
| score -= 20 |
| reasons.append(f"非工作时间({hour}点)") |
| if dt.weekday() >= 5: |
| score -= 10 |
| reasons.append("周末") |
| else: |
| score += 5 |
| reasons.append("工作日") |
| if dt.minute in {0, 30}: |
| score += 10 |
| reasons.append("整点/半点") |
| elif dt.minute in {15, 45}: |
| score += 5 |
| reasons.append("刻钟") |
| score = max(0, min(100, score)) |
| return score > 30, score, "; ".join(reasons) |
|
|
| def _validate_regex_results(self, regex_results: list[tuple[datetime, str, int | float]]) -> list[dict[str, Any]]: |
| """对正则结果进行合理性校验和综合评分""" |
| validated: list[dict[str, Any]] = [] |
| for dt, matched_text, context_score in regex_results: |
| is_valid, validity_score, validity_reason = self._validate_hearing_datetime(dt) |
| combined_score = context_score * 0.6 + validity_score * 0.4 |
| validated.append( |
| { |
| "datetime": dt, |
| "matched_text": matched_text, |
| "context_score": context_score, |
| "validity_score": validity_score, |
| "validity_reason": validity_reason, |
| "combined_score": combined_score, |
| "is_valid": is_valid, |
| } |
| ) |
| logger.debug( |
| f"正则候选: {dt}, 上下文={context_score}, 合理性={validity_score}({validity_reason}), " |
| f"综合={combined_score:.1f}, 有效={is_valid}" |
| ) |
| return validated |
|
|
| def _select_best_from_conflict( |
| self, |
| best_regex_dt: datetime, |
| best_regex_combined: float, |
| ollama_datetime: datetime, |
| ollama_validity_score: int, |
| valid_regex_results: list[dict[str, Any]], |
| ) -> tuple[datetime | None, str]: |
| """处理正则与 Ollama 结果冲突""" |
| date_diff = abs((best_regex_dt.date() - ollama_datetime.date()).days) |
| time_diff = abs((best_regex_dt - ollama_datetime).total_seconds()) |
| if date_diff == 0 and time_diff < 3600: |
| logger.info(f"正则和Ollama结果一致: {best_regex_dt}") |
| return best_regex_dt, "regex+ollama(一致)" |
| if date_diff == 0: |
| logger.warning(f"时间冲突: 正则={best_regex_dt}, Ollama={ollama_datetime}, 时间差={time_diff}秒") |
| if best_regex_combined >= 40: |
| return best_regex_dt, f"regex(score={best_regex_combined:.0f},时间冲突)" |
| return ollama_datetime, f"ollama(validity={ollama_validity_score},时间冲突,正则得分低)" |
| logger.warning(f"日期冲突: 正则={best_regex_dt.date()}, Ollama={ollama_datetime.date()}, 差异={date_diff}天") |
| for result in valid_regex_results[1:]: |
| result_dt = cast(datetime, result["datetime"]) |
| if abs((result_dt.date() - ollama_datetime.date()).days) == 0: |
| logger.info(f"找到与Ollama一致的备选正则结果: {result['datetime']}") |
| return result_dt, f"regex(score={result['combined_score']:.0f},与ollama一致)" |
| if best_regex_combined >= 60: |
| return best_regex_dt, f"regex(score={best_regex_combined:.0f},日期冲突)" |
| if ollama_validity_score >= 50: |
| return ollama_datetime, f"ollama(validity={ollama_validity_score},日期冲突)" |
| if best_regex_combined >= ollama_validity_score: |
| return best_regex_dt, f"regex(score={best_regex_combined:.0f},低置信度)" |
| return ollama_datetime, f"ollama(validity={ollama_validity_score},低置信度)" |
|
|
| def _select_best_datetime( |
| self, regex_results: list[tuple[datetime, str, int | float]], ollama_datetime: datetime | None |
| ) -> tuple[datetime | None, str]: |
| """从正则提取结果和 Ollama 结果中选择最佳时间""" |
| if not regex_results and not ollama_datetime: |
| return None, "无法提取" |
| validated_regex_results = self._validate_regex_results(regex_results) |
| valid_regex_results = sorted( |
| [r for r in validated_regex_results if r["is_valid"]], |
| key=lambda x: float(x["combined_score"]), |
| reverse=True, |
| ) |
| ollama_valid = False |
| ollama_validity_score = 0 |
| ollama_validity_reason = "" |
| if ollama_datetime: |
| ollama_valid, ollama_validity_score, ollama_validity_reason = self._validate_hearing_datetime( |
| ollama_datetime |
| ) |
| logger.debug( |
| f"Ollama候选: {ollama_datetime}," |
| f" 合理性={ollama_validity_score}({ollama_validity_reason}), 有效={ollama_valid}" |
| ) |
| if not valid_regex_results: |
| if ollama_datetime and ollama_valid: |
| logger.info(f"仅使用 Ollama 结果: {ollama_datetime} (合理性={ollama_validity_score})") |
| return ollama_datetime, f"ollama(validity={ollama_validity_score})" |
| if ollama_datetime: |
| logger.warning(f"Ollama 结果合理性较低: {ollama_datetime} ({ollama_validity_reason})") |
| return ollama_datetime, f"ollama(低合理性:{ollama_validity_reason})" |
| if validated_regex_results: |
| best_invalid = max(validated_regex_results, key=lambda x: float(x["combined_score"])) |
| logger.warning( |
| f"所有正则结果合理性较低，使用最佳候选:" |
| f" {best_invalid['datetime']} ({best_invalid['validity_reason']})" |
| ) |
| return ( |
| cast(datetime, best_invalid["datetime"]), |
| f"regex(低合理性:{best_invalid['validity_reason']})", |
| ) |
| return None, "无法提取" |
| best_regex = valid_regex_results[0] |
| best_regex_dt = cast(datetime, best_regex["datetime"]) |
| best_regex_combined = float(best_regex["combined_score"]) |
| if not ollama_datetime or not ollama_valid: |
| logger.info(f"使用正则结果: {best_regex_dt}, 综合得分={best_regex_combined:.1f}") |
| return best_regex_dt, f"regex(score={best_regex_combined:.0f})" |
| return self._select_best_from_conflict( |
| best_regex_dt, best_regex_combined, ollama_datetime, ollama_validity_score, valid_regex_results |
| ) |
|
|
