Fix Build-in metadata not working (#788) (#16748)

rag/svr/task_executor_refactor/chunk_post_processor.py

@@ -28,6 +28,7 @@ import json
 import logging
 import random
 import re
+from datetime import datetime
 from timeit import default_timer as timer
 from typing import Dict, List
 
@@ -221,6 +222,28 @@ async def generate_metadata(docs: List[Dict], ctx: TaskContext) -> None:
         ctx.progress_cb(msg="Metadata generation {} chunks completed in {:.2f}s".format(len(docs), timer() - st))
 
 
+def apply_built_in_metadata(ctx: TaskContext) -> None:
+    built_in_meta_config = ctx.parser_config.get("built_in_metadata", [])
+    if not built_in_meta_config:
+        return
+
+    built_in_meta = {}
+    for item in built_in_meta_config:
+        key = item.get("key", "")
+        if key == "update_time":
+            built_in_meta["update_time"] = str(datetime.now()).replace("T", " ")[:19]
+        elif key == "file_name":
+            built_in_meta["file_name"] = ctx.name
+    if built_in_meta:
+        existing_meta = DocMetadataService.get_document_metadata(ctx.doc_id)
+        existing_meta = existing_meta if isinstance(existing_meta, dict) else {}
+        existing_meta = update_metadata_to(existing_meta, built_in_meta)
+        if ctx.write_interceptor:
+            ctx.write_interceptor.intercept("DocMetadataService.update_document_metadata")
+        else:
+            DocMetadataService.update_document_metadata(ctx.doc_id, existing_meta)
+
+
 async def apply_tags(docs: List[Dict], ctx: TaskContext) -> None:
     """Apply tags to chunks.
 

rag/svr/task_executor_refactor/chunk_service.py

@@ -55,6 +55,7 @@ from rag.svr.task_executor_refactor.chunk_post_processor import (
     extract_keywords,
     generate_questions,
     generate_metadata,
+    apply_built_in_metadata,
     apply_tags,
 )
 
@@ -159,6 +160,7 @@ class ChunkService:
 
         if ctx.parser_config.get("enable_metadata", False) and (ctx.parser_config.get("metadata") or ctx.parser_config.get("built_in_metadata")):
             await generate_metadata(docs, ctx)
+            apply_built_in_metadata(ctx)
         metadata_list = [d for d in docs if d.get("metadata_obj")]
         self._task_context.recording_context.record("metadata_list_generated", metadata_list)