1use crate::{graph::{self, *}, memory::{self, *}, notes::{self, NoteFileInput},
3 schema, storage::{self, KnowledgeBase}, text, types::*, Error, Result};
4use rusqlite::{params, types::ValueRef, Connection, OpenFlags, OptionalExtension};
5use serde::{Deserialize, Serialize};
6use serde_json::{json, Value};
7use std::{collections::{BTreeMap, BTreeSet, HashMap}, path::{Path, PathBuf}};
8
9#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
10#[serde(rename_all = "snake_case")]
11pub enum LegacySource { #[serde(rename = "p_ai")] Pai, WorldTree, AngelMemory }
12fn dry_run() -> bool { true }
13#[derive(Debug, Clone, Serialize, Deserialize)]
14pub struct ImportRequest {
15 pub source: LegacySource,
16 pub source_id: String,
18 pub database: PathBuf,
19 pub destination: PathBuf,
20 #[serde(default)] pub graph_database: Option<PathBuf>,
21 #[serde(default)] pub lookup_database: Option<PathBuf>,
22 #[serde(default)] pub notes_root: Option<PathBuf>,
24 #[serde(default = "default_namespace")] pub namespace: String,
25 #[serde(default = "public_scope")] pub scope: String,
26 #[serde(default = "dry_run")] pub dry_run: bool,
27}
28#[derive(Debug, Clone, Serialize, Deserialize)]
30pub struct IdMapping { pub source_table: String, pub source_id: String, pub target_id: i64 }
31#[derive(Debug, Clone, Serialize, Deserialize)]
32pub struct ImportReport {
33 pub source_id: String, pub source_fingerprint: String,
34 pub applied: bool, pub already_imported: bool,
35 pub counts: BTreeMap<String,usize>, pub id_map: Vec<IdMapping>,
36 pub conflicts: Vec<String>, pub missing_sources: Vec<String>, pub warnings: Vec<String>,
37 pub index_ready: bool, pub index_error: Option<String>,
38}
39
40fn key(table: &str, old: &str) -> String { format!("{table}\u{1f}{old}") }
42
43#[derive(Serialize)]
44struct MemoryDraft { table: String, source_id: String, input: MemoryInput }
45#[derive(Serialize)]
46struct EntityDraft { table: String, source_id: String, input: EntityInput }
47impl EntityDraft { fn key(&self) -> String { key(&self.table, &self.source_id) } }
48#[derive(Serialize)]
49struct RelationDraft { table: String, source_id: String, record: RecordInput, subject: String, predicate: String, object: String, confidence: f64, reason: String }
50#[derive(Serialize)]
51struct EventDraft { table: String, source_id: String, record: RecordInput, name: String, summary: String, participants: Vec<String>, confidence: f64, reason: String }
52#[derive(Serialize)]
53struct NoteDraft { table: String, source_id: String, input: NoteFileInput }
54#[derive(Default, Serialize)]
55struct Plan { memories: Vec<MemoryDraft>, entities: Vec<EntityDraft>, relations: Vec<RelationDraft>, events: Vec<EventDraft>,
56 notes: Vec<NoteDraft>, roots: BTreeMap<String, String>, missing: BTreeSet<String>, warnings: Vec<String> }
57type Row = serde_json::Map<String,Value>;
58
59fn source_conn(path: &Path) -> Result<Connection> {
60 let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX)?;
61 conn.execute_batch("PRAGMA query_only=ON; BEGIN DEFERRED;")?;
62 Ok(conn)
63}
64fn has_table(conn: &Connection, name: &str) -> Result<bool> {
65 Ok(conn.query_row("SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1)", [name], |r|r.get(0))?)
66}
67fn rows(conn: &Connection, table: &str) -> Result<Vec<Row>> {
68 if !table.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') { return Err(Error::Validation("invalid source table".into())); }
69 if !has_table(conn,table)? { return Ok(vec![]); }
70 let mut stmt=conn.prepare(&format!("SELECT * FROM {table} ORDER BY 1"))?;
71 let names:Vec<_>=stmt.column_names().into_iter().map(String::from).collect();
72 let mut query=stmt.query([])?;let mut result=Vec::new();
73 while let Some(row)=query.next()? {
74 let mut object=Row::new();
75 for (i,name) in names.iter().enumerate() {
76 let value=match row.get_ref(i)? {
77 ValueRef::Null=>Value::Null,ValueRef::Integer(v)=>json!(v),ValueRef::Real(v)=>json!(v),
78 ValueRef::Text(v)=>Value::String(String::from_utf8(v.to_vec()).map_err(|_|Error::Validation(format!("non-UTF-8 source in {table}.{name}")))?),
79 ValueRef::Blob(v)=>json!({"bytes":v}),
80 };
81 object.insert(name.clone(),value);
82 }
83 result.push(object);
84 }
85 Ok(result)
86}
87fn string(row:&Row,key:&str)->String {
88 match row.get(key) { Some(Value::String(s))=>s.clone(),Some(Value::Number(n))=>n.to_string(),_=>String::new() }
89}
90fn strings(value:Option<&Value>)->Vec<String> {
91 match value { Some(Value::Array(a))=>a.iter().filter_map(|v|v.as_str().map(String::from)).collect(),Some(Value::String(s))=>vec![s.clone()],_=>vec![] }
92}
93fn number(row:&Row,key:&str,default:f64)->f64 { row.get(key).and_then(Value::as_f64).unwrap_or(default) }
94fn timestamp(row:&Row,key:&str)->Result<Option<i64>> {
95 let Some(value)=row.get(key).filter(|v|!v.is_null()) else {return Ok(None)};
96 if let Some(s)=value.as_str() {
97 if s.trim().is_empty(){return Ok(None)}
98 if let Ok(dt)=chrono::DateTime::parse_from_rfc3339(s){return Ok(Some(dt.timestamp_micros()))}
99 if let Ok(dt)=chrono::NaiveDateTime::parse_from_str(s.trim(),"%Y-%m-%dT%H:%M:%S%.f"){return Ok(Some(dt.and_utc().timestamp_micros()))}
100 if let Ok(seconds)=s.parse::<f64>(){return seconds_to_us(seconds).map(Some)}
101 }
102 if let Some(seconds)=value.as_f64(){return seconds_to_us(seconds).map(Some)}
103 Err(Error::Validation(format!("invalid source timestamp {key}: {value}")))
104}
105fn seconds_to_us(seconds:f64)->Result<i64>{
106 let value=seconds*1_000_000.0;
107 if !value.is_finite() || value < i64::MIN as f64 || value >= i64::MAX as f64 {return Err(Error::Validation("source timestamp out of range".into()))}
108 Ok(value.round() as i64)
109}
110fn metadata(row:&Row)->Result<Row>{
111 match row.get("metadata_json") {
112 Some(Value::String(s)) if !s.trim().is_empty()=>Ok(serde_json::from_str(s)?),
113 _=>Ok(Row::new()),
114 }
115}
116fn namespace(req:&ImportRequest,domain:&str)->String {
117 if matches!(req.source,LegacySource::WorldTree) && !domain.trim().is_empty(){format!("{}/{}",req.namespace,domain.trim())}else{req.namespace.clone()}
118}
119fn record(req:&ImportRequest,table:&str,old:&str,row:&Row,domain:&str)->Result<RecordInput>{
121 if old.is_empty(){return Err(Error::Validation(format!("missing source ID in {table}")))}
122 let namespace=namespace(req,domain);
123 let mut metadata=metadata(row)?;
124 metadata.insert("legacy".into(),json!({"format":req.source,"source_id":req.source_id,"table":table,"id":old,"row":row}));
125 let scope=string(row,"memory_scope");
126 Ok(RecordInput{id:None,namespace,scope:if scope.trim().is_empty(){req.scope.clone()}else{scope},metadata,
127 created_at_us:Some(timestamp(row,"created_at")?.unwrap_or(0)),
128 updated_at_us:Some(timestamp(row,"updated_at")?.or(timestamp(row,"created_at")?).unwrap_or(0)),..Default::default()})
129}
130fn tag_map(conn:&Connection,tags:&str,relations:&str,id_col:&str)->Result<BTreeMap<String,Vec<String>>>{
131 let names:BTreeMap<_,_>=rows(conn,tags)?.iter().map(|r|(string(r,"id"),string(r,"name"))).collect();
132 let mut map:BTreeMap<String,Vec<String>>=BTreeMap::new();
133 for row in rows(conn,relations)? {if let Some(name)=names.get(&string(&row,"tag_id")){map.entry(string(&row,id_col)).or_default().push(name.clone());}}
134 Ok(map)
135}
136fn evidence(value:&Value,fallback:&Row)->Vec<Evidence>{
137 let values=match value {Value::Array(a)=>a.clone(),Value::Object(_)=>vec![value.clone()],_=>vec![]};
138 let mut result=Vec::new();
139 for item in values {
140 if let Some(row)=item.as_object(){
141 let source=string(row,"source");let source=if source.is_empty(){string(row,"file_path")}else{source};
142 if source.is_empty(){continue}
143 let start=row.get("line_start").and_then(Value::as_u64).filter(|v|*v>0).map(|v|v as usize);
144 let end=row.get("line_end").and_then(Value::as_u64).map(|v|v as usize);
145 let range=start.zip(end).filter(|(s,e)|e>=s);
146 result.push(Evidence{source,source_revision:None,chunk_id:None,
147 offset:range.map(|v|v.0),limit:range.map(|(s,e)|e-s+1),quote:string(row,"quote"),metadata:row.clone()});
148 }
149 }
150 if result.is_empty(){
151 let source=string(fallback,"file_path");
152 if !source.is_empty(){result.push(Evidence{source,..Default::default()});}
153 }
154 result
155}
156fn memory_rows(req:&ImportRequest,plan:&mut Plan,conn:&Connection)->Result<Vec<Row>>{
157 let (table,tags,rel)=match req.source{LegacySource::Pai=>("memory_record","global_tag","memory_tag_rel"),LegacySource::AngelMemory=>("memory_records","global_tags","memory_tag_rel"),LegacySource::WorldTree=>("world_tree_memory","world_tree_tag","world_tree_memory_tag")};
158 if !has_table(conn,table)? {return Err(Error::Validation(format!("expected legacy table {table}")))}
159 let memory_tags=tag_map(conn,tags,rel,"memory_id")?;
160 for row in rows(conn,table)? {
161 let old=string(&row,"id");let metadata=metadata(&row)?;
162 let mut rec=record(req,table,&old,&row,&string(&metadata,"domain"))?;
163 rec.tags=memory_tags.get(&old).cloned().unwrap_or_default();
164 rec.evidence=evidence(metadata.get("evidence").unwrap_or(&Value::Null),&metadata);
165 let kind=string(&row,"memory_type");
166 let memory_type=match kind.as_str(){"知识记忆"=>"knowledge","事件记忆"=>"event","技能记忆"=>"skill","任务记忆"=>"task","情感记忆"=>"emotion",""=>"knowledge",_=>&kind}.to_string();
167 let state=MemoryState{pinned:row.get("is_active").is_some_and(|v|v.as_bool()==Some(true)||v.as_i64()==Some(1)),
168 strength:row.get("strength").and_then(Value::as_i64).unwrap_or(1),useful_count:row.get("useful_count").and_then(Value::as_i64).unwrap_or(0),useful_score:number(&row,"useful_score",0.0),
169 last_recalled_at_us:timestamp(&row,"last_recalled_at")?.filter(|v|*v!=0),last_decay_at_us:timestamp(&row,"last_decay_at")?.filter(|v|*v!=0)};
170 plan.memories.push(MemoryDraft{table:table.into(),source_id:old.clone(),
171 input:MemoryInput{record:rec,memory_type,judgment:string(&row,"judgment"),reasoning:string(&row,"reasoning"),state:Some(state)}});
172 }
173 let note_table=if matches!(req.source,LegacySource::Pai){"note_index_record"}else{"note_index_records"};
174 let mut notes=rows(conn,note_table)?;
175 if matches!(req.source,LegacySource::Pai){
176 let tags=tag_map(conn,"global_tag","note_tag_rel","source_id")?;
177 for row in &mut notes {row.insert("tags".into(),json!(tags.get(&string(row,"source_id")).cloned().unwrap_or_default()));}
178 }
179 Ok(notes)
180}
181fn raw_graph(req:&ImportRequest,plan:&mut Plan)->Result<Vec<Row>>{
182 let Some(path)=&req.graph_database else{return Ok(vec![])};
183 let conn=source_conn(path)?;
184 if !has_table(&conn,"world_tree_graph")?{return Err(Error::Validation("expected world_tree_graph table".into()))}
185 let rows=rows(&conn,"world_tree_graph")?;
186 let tags=tag_map(&conn,"world_tree_graph_tag","world_tree_graph_tag_map","graph_id")?;
187 for row in &rows {
188 let metadata=metadata(row)?;let old=string(row,"id");
189 let mut rec=record(req,"world_tree_graph",&old,row,&string(&metadata,"domain"))?;
190 rec.tags=tags.get(&old).cloned().unwrap_or_default();rec.tags.push("legacy:graph-extraction".into());
191 rec.evidence=evidence(metadata.get("evidence").unwrap_or(&Value::Null),&metadata);
192 plan.memories.push(MemoryDraft{table:"world_tree_graph".into(),source_id:old.clone(),
193 input:MemoryInput{record:rec,memory_type:"graph_extraction".into(),judgment:string(row,"judgment"),reasoning:string(row,"reasoning"),state:None}});
194 }
195 Ok(rows)
196}
197fn lookup_graph(req:&ImportRequest,plan:&mut Plan,path:&Path)->Result<()> {
198 let conn=source_conn(path)?;
199 if !has_table(&conn,"entities")?{return Err(Error::Validation("expected entities table in lookup database".into()))}
200 let mut aliases:BTreeMap<String,Vec<String>>=BTreeMap::new();
201 for row in rows(&conn,"entity_aliases")?{aliases.entry(string(&row,"entity_id")).or_default().push(string(&row,"term"));}
202 let mut attributes:BTreeMap<String,BTreeMap<String,Vec<String>>>=BTreeMap::new();
203 for row in rows(&conn,"entity_attributes")?{attributes.entry(string(&row,"entity_id")).or_default().entry(string(&row,"attr_key")).or_default().push(string(&row,"attr_value"));}
204 for row in rows(&conn,"entities")? {
205 let old=string(&row,"entity_id");let rec=record(req,"entities",&old,&row,&string(&row,"domain"))?;
206 plan.entities.push(EntityDraft{table:"entities".into(),source_id:old.clone(),
207 input:EntityInput{record:rec,name:string(&row,"canonical_name"),entity_type:{let t=string(&row,"entity_type");if t.is_empty(){"concept".into()}else{t}},
208 summary:string(&row,"summary"),aliases:aliases.remove(&old).unwrap_or_default(),attributes:attributes.remove(&old).unwrap_or_default()}});
209 }
210 for row in rows(&conn,"relations")?{
211 let old=string(&row,"relation_id");let rec=record(req,"relations",&old,&row,&string(&row,"domain"))?;
212 plan.relations.push(RelationDraft{table:"relations".into(),source_id:old,record:rec,
213 subject:key("entities",&string(&row,"subject_entity_id")),object:key("entities",&string(&row,"object_entity_id")),
214 predicate:string(&row,"predicate"),reason:string(&row,"reason"),confidence:number(&row,"confidence",0.8)});
215 }
216 let mut participants:BTreeMap<String,Vec<String>>=BTreeMap::new();
217 for row in rows(&conn,"event_participants")?{participants.entry(string(&row,"event_id")).or_default().push(key("entities",&string(&row,"entity_id")));}
218 let mut event_aliases:BTreeMap<String,Vec<String>>=BTreeMap::new();
219 for row in rows(&conn,"event_aliases")?{event_aliases.entry(string(&row,"event_id")).or_default().push(string(&row,"term"));}
220 for row in rows(&conn,"events")?{
221 let old=string(&row,"event_id");let mut rec=record(req,"events",&old,&row,&string(&row,"domain"))?;
222 rec.metadata.insert("legacy_event_aliases".into(),json!(event_aliases.remove(&old).unwrap_or_default()));
223 plan.events.push(EventDraft{table:"events".into(),source_id:old.clone(),record:rec,name:string(&row,"event_name"),
224 summary:string(&row,"summary"),reason:string(&row,"reason"),confidence:number(&row,"confidence",0.8),
225 participants:participants.remove(&old).unwrap_or_default()});
226 }
227 Ok(())
228}
229fn attributes(value:Option<&Value>)->BTreeMap<String,Vec<String>>{
230 value.and_then(Value::as_object).map(|m|m.iter().map(|(k,v)|(k.clone(),match v {Value::String(s)=>vec![s.clone()],Value::Array(a)=>a.iter().map(|v|v.as_str().map(String::from).unwrap_or_else(||v.to_string())).collect(),_=>vec![v.to_string()]})).collect()).unwrap_or_default()
231}
232fn extraction_graph(req:&ImportRequest,plan:&mut Plan,raw:&[Row])->Result<()> {
233 if !raw.is_empty(){plan.warnings.push("No canonical lookup database supplied: extraction-local entities remain separate across source records.".into());}
234 for row in raw {
235 let metadata=metadata(row)?;let old=string(row,"id");let domain=string(&metadata,"domain");
236 let mut local:BTreeMap<String,Vec<String>>=BTreeMap::new();
237 for (i,item) in metadata.get("entities").and_then(Value::as_array).into_iter().flatten().enumerate(){
238 let Some(item)=item.as_object()else{continue};let name=string(item,"name");
239 let sid=format!("{old}:{i}");
240 let mut rec=record(req,"extracted_entities",&sid,row,&domain)?;
241 rec.metadata.insert("extraction".into(),json!(item));rec.evidence=evidence(item.get("evidence").unwrap_or(&Value::Null),&metadata);
242 let entity_key=key("extracted_entities",&sid);let aliases=strings(item.get("aliases"));
243 for alias in std::iter::once(&name).chain(aliases.iter()){let ids=local.entry(text::normalized_tag(alias)).or_default();if !ids.contains(&entity_key){ids.push(entity_key.clone());}}
244 let kind=string(item,"type");plan.entities.push(EntityDraft{table:"extracted_entities".into(),source_id:sid,
245 input:EntityInput{record:rec,name,entity_type:if kind.is_empty(){"concept".into()}else{kind},aliases,attributes:attributes(item.get("attributes")),summary:string(item,"summary")}});
246 }
247 let resolve=|name:&str,plan:&mut Plan,local:&mut BTreeMap<String,Vec<String>>|->Result<String>{
249 let name=name.trim();if name.is_empty(){return Err(Error::Validation(format!("empty entity reference in graph {old}")))}
250 if let Some(ids)=local.get(&text::normalized_tag(name)){
251 if ids.len()!=1{return Err(Error::Conflict(format!("ambiguous entity {name} in graph {old}")))}return Ok(ids[0].clone())
252 }
253 let sid=format!("{old}:{name}");
254 let mut rec=record(req,"inferred_entities",&sid,row,&domain)?;
255 rec.metadata.insert("inferred_from_legacy_reference".into(),json!(true));
256 let entity_key=key("inferred_entities",&sid);local.insert(text::normalized_tag(name),vec![entity_key.clone()]);
257 plan.entities.push(EntityDraft{table:"inferred_entities".into(),source_id:sid,
258 input:EntityInput{record:rec,name:name.into(),entity_type:"concept".into(),aliases:vec![],attributes:BTreeMap::new(),summary:String::new()}});Ok(entity_key)
259 };
260 for (i,item) in metadata.get("relations").and_then(Value::as_array).into_iter().flatten().enumerate(){
261 let Some(item)=item.as_object()else{continue};
262 let subject=resolve(&string(item,"subject"),plan,&mut local)?;let object=resolve(&string(item,"object"),plan,&mut local)?;
263 let sid=format!("{old}:{i}");
264 let mut rec=record(req,"extracted_relations",&sid,row,&domain)?;
265 rec.metadata.insert("extraction".into(),json!(item));rec.evidence=evidence(item.get("evidence").unwrap_or(&Value::Null),&metadata);
266 plan.relations.push(RelationDraft{table:"extracted_relations".into(),source_id:sid,record:rec,subject,object,
267 predicate:string(item,"predicate"),reason:string(item,"reason"),confidence:number(item,"confidence",0.8)});
268 }
269 for (i,item) in metadata.get("events").and_then(Value::as_array).into_iter().flatten().enumerate(){
270 let Some(item)=item.as_object()else{continue};let mut participants=Vec::new();
271 for name in strings(item.get("participants")){participants.push(resolve(&name,plan,&mut local)?);}
272 let sid=format!("{old}:{i}");
273 let mut rec=record(req,"extracted_events",&sid,row,&domain)?;
274 rec.metadata.insert("extraction".into(),json!(item));rec.evidence=evidence(item.get("evidence").unwrap_or(&Value::Null),&metadata);
275 plan.events.push(EventDraft{table:"extracted_events".into(),source_id:sid,record:rec,name:string(item,"name"),
276 summary:string(item,"summary"),participants,confidence:number(item,"confidence",0.8),reason:string(item,"reason")});
277 }
278 }
279 Ok(())
280}
281
282fn attach_evidence(req:&ImportRequest,plan:&mut Plan,raw:&[Row])->Result<()> {
285 let mut aliases:BTreeMap<(String,String),BTreeSet<String>>=BTreeMap::new();
286 for entity in &plan.entities{for name in std::iter::once(&entity.input.name).chain(entity.input.aliases.iter()){
287 aliases.entry((entity.input.record.namespace.clone(),text::normalized_tag(name))).or_default().insert(entity.key());}}
288 let entity_index:HashMap<String,usize>=plan.entities.iter().enumerate().map(|(i,e)|(e.key().to_owned(),i)).collect();
289 let mut relation_index:HashMap<(String,String,String,String),Vec<usize>>=HashMap::new();
290 for (i,r) in plan.relations.iter().enumerate(){relation_index.entry((r.record.namespace.clone(),r.subject.clone(),r.object.clone(),r.predicate.clone())).or_default().push(i);}
291 let mut event_index:HashMap<(String,String),Vec<usize>>=HashMap::new();
292 for (i,e) in plan.events.iter().enumerate(){event_index.entry((e.record.namespace.clone(),e.name.clone())).or_default().push(i);}
293 for row in raw{
294 let meta=metadata(row)?;let ns=namespace(req,&string(&meta,"domain"));
295 let resolve=|name:&str|->Option<String>{let ids=aliases.get(&(ns.clone(),text::normalized_tag(name)))?;if ids.len()==1{ids.first().cloned()}else{None}};
296 for item in meta.get("entities").and_then(Value::as_array).into_iter().flatten(){
297 let Some(item)=item.as_object()else{continue};let Some(id)=resolve(&string(item,"name"))else{continue};
298 if let Some(&i)=entity_index.get(&id){plan.entities[i].input.record.evidence.extend(evidence(item.get("evidence").unwrap_or(&Value::Null),&meta));}
299 }
300 for item in meta.get("relations").and_then(Value::as_array).into_iter().flatten(){
301 let Some(item)=item.as_object()else{continue};let (Some(s),Some(o))=(resolve(&string(item,"subject")),resolve(&string(item,"object")))else{continue};
302 if let Some(idx)=relation_index.get(&(ns.clone(),s,o,string(item,"predicate"))){for &i in idx{plan.relations[i].record.evidence.extend(evidence(item.get("evidence").unwrap_or(&Value::Null),&meta));}}
303 }
304 for item in meta.get("events").and_then(Value::as_array).into_iter().flatten(){
305 let Some(item)=item.as_object()else{continue};
306 if let Some(idx)=event_index.get(&(ns.clone(),string(item,"name"))){for &i in idx{plan.events[i].record.evidence.extend(evidence(item.get("evidence").unwrap_or(&Value::Null),&meta));}}
307 }
308 }
309 Ok(())
310}
311fn collect_files(root:&Path)->Result<Vec<PathBuf>>{
312 let mut result=Vec::new();let mut stack=vec![root.to_path_buf()];
313 while let Some(path)=stack.pop(){
314 for entry in std::fs::read_dir(path)?{let entry=entry?;let kind=entry.file_type()?;
315 if kind.is_symlink(){continue}if kind.is_dir(){stack.push(entry.path())}else if kind.is_file(){
316 let ext=entry.path().extension().and_then(|s|s.to_str()).unwrap_or("").to_lowercase();if matches!(ext.as_str(),"md"|"markdown"|"txt"){result.push(entry.path());}
317 }
318 }
319 }
320 result.sort();Ok(result)
321}
322fn import_notes(req:&ImportRequest,plan:&mut Plan,index_rows:Vec<Row>,raw:&[Row])->Result<()> {
323 let root=req.notes_root.as_ref().map(std::fs::canonicalize).transpose()?;
324 let mut sources:BTreeMap<String,Row>=BTreeMap::new();
325 for row in index_rows {let source=string(&row,"source_file_path");if !source.is_empty(){sources.insert(source,row);}}
326 for row in raw {let meta=metadata(row)?;let source=string(&meta,"file_path");if !source.is_empty(){sources.entry(source).or_insert(meta);}}
327 if let Some(root)=&root {
328 for file in collect_files(root)? {let relative=file.strip_prefix(root).map_err(|e|Error::Validation(e.to_string()))?.to_string_lossy().replace('\\',"/");sources.entry(relative).or_default();}
329 }
330 let mut seen=BTreeSet::new();
331 for (source,row) in sources {
332 let Some(root)=&root else{plan.missing.insert(source);continue};
333 let candidate=root.join(source.replace('\\',"/"));
334 let path=match std::fs::canonicalize(&candidate){Ok(p)=>p,Err(e) if e.kind()==std::io::ErrorKind::NotFound=>{plan.missing.insert(source);continue},Err(e)=>return Err(e.into())};
335 if !path.starts_with(root){plan.missing.insert(source.clone());plan.warnings.push(format!("Skipped note outside notes_root: {source}"));continue}
336 let relative=path.strip_prefix(root).map_err(|e|Error::Validation(e.to_string()))?.to_string_lossy().replace('\\',"/");
337 let domain=if matches!(req.source,LegacySource::WorldTree){relative.split('/').next().filter(|_|relative.contains('/')).unwrap_or("")}else{""};
338 let domain={let explicit=string(&row,"domain");if explicit.is_empty(){domain.to_string()}else{explicit}};
339 let ns=namespace(req,&domain);
340 plan.roots.entry(ns.clone()).or_insert_with(|| root.to_string_lossy().replace('\\',"/"));
342 if !seen.insert((ns.clone(),path.clone())){continue}
343 let sid=format!("{ns}:{relative}");
344 let mut rec=record(req,"notes",&sid,&row,&domain)?;
345 rec.tags=strings(row.get("tags"));
346 let note=NoteFileInput{record:rec,path:path.clone(),chunk_chars:220};
347 plan.notes.push(NoteDraft{table:"notes".into(),source_id:sid,input:note});
348 }
349 Ok(())
350}
351fn build(req:&ImportRequest)->Result<Plan>{
352 storage::validate_identity("source_id",&req.source_id)?;storage::validate_identity("namespace",&req.namespace)?;storage::validate_identity("scope",&req.scope)?;
353 if !matches!(req.source,LegacySource::WorldTree)&&(req.graph_database.is_some()||req.lookup_database.is_some()){return Err(Error::Validation("graph_database and lookup_database apply only to world_tree".into()))}
354 let conn=source_conn(&req.database)?;let mut plan=Plan::default();let note_rows=memory_rows(req,&mut plan,&conn)?;
355 let raw=raw_graph(req,&mut plan)?;
356 if let Some(path)=&req.lookup_database{lookup_graph(req,&mut plan,path)?;attach_evidence(req,&mut plan,&raw)?;}else{extraction_graph(req,&mut plan,&raw)?;}
357 import_notes(req,&mut plan,note_rows,&raw)?;
358 plan.warnings.push("Legacy Tantivy/FAISS/vector caches are not copied. Vectors are generated inside the library: register an embedder and run embeddings.sync to fill them.".into());
359 Ok(plan)
360}
361fn apply(conn:&Connection,plan:&Plan,mappings:&mut Vec<IdMapping>,conflicts:&mut Vec<String>,documents:&mut Vec<crate::index::IndexDocument>)->Result<()> {
363 for draft in &plan.memories { let (memory,document)=memory::upsert(conn,&draft.input)?; documents.push(document);
364 mappings.push(IdMapping{source_table:draft.table.clone(),source_id:draft.source_id.clone(),target_id:memory.header.id}); }
365 let mut map:HashMap<String,i64>=HashMap::new();
366 for draft in &plan.entities {
367 let (entity,document)=graph::upsert_entity(conn,&draft.input)?; documents.push(document);
368 map.insert(draft.key(),entity.header.id);
369 mappings.push(IdMapping{source_table:draft.table.clone(),source_id:draft.source_id.clone(),target_id:entity.header.id});
370 }
371 for draft in &plan.relations {
372 let (Some(&subject),Some(&object))=(map.get(&draft.subject),map.get(&draft.object)) else {
373 conflicts.push(format!("skipped relation {}: unresolved entity reference {} -> {}",draft.source_id,draft.subject,draft.object));continue;};
374 let (relation,document)=graph::upsert_relation(conn,&RelationInput{record:draft.record.clone(),subject_id:subject,predicate:draft.predicate.clone(),object_id:object,confidence:draft.confidence,reason:draft.reason.clone()})?;
375 documents.push(document);
376 mappings.push(IdMapping{source_table:draft.table.clone(),source_id:draft.source_id.clone(),target_id:relation.header.id});
377 }
378 for draft in &plan.events {
379 let mut participants=Vec::new();
380 for participant in &draft.participants { match map.get(participant){Some(id)=>participants.push(*id),None=>conflicts.push(format!("event {}: dropped unresolved participant {participant}",draft.source_id))} }
381 let (event,document)=graph::upsert_event(conn,&EventInput{record:draft.record.clone(),name:draft.name.clone(),summary:draft.summary.clone(),participants,confidence:draft.confidence,reason:draft.reason.clone()})?;
382 documents.push(document);
383 mappings.push(IdMapping{source_table:draft.table.clone(),source_id:draft.source_id.clone(),target_id:event.header.id});
384 }
385 for (namespace, root) in &plan.roots {
387 let namespace_id = storage::term_id(conn, namespace)?;
388 conn.execute("INSERT INTO namespace_roots(namespace_id,root) VALUES (?1,?2) ON CONFLICT(namespace_id) DO UPDATE SET root=excluded.root",
389 params![namespace_id, root])?;
390 }
391 for draft in &plan.notes { let (note,staged)=notes::sync_file(conn,&draft.input)?; documents.extend(staged);
392 mappings.push(IdMapping{source_table:draft.table.clone(),source_id:draft.source_id.clone(),target_id:note.header.id}); }
393 Ok(())
394}
395pub fn import_legacy(req:&ImportRequest)->Result<ImportReport>{
399 let plan=build(req)?;
400 let fingerprint=text::digest(&serde_json::to_string(&plan)?);
401 let mut report=ImportReport{source_id:req.source_id.clone(),source_fingerprint:fingerprint.clone(),applied:false,already_imported:false,
402 counts:BTreeMap::new(),id_map:vec![],conflicts:vec![],missing_sources:plan.missing.iter().cloned().collect(),warnings:plan.warnings.clone(),index_ready:false,index_error:None};
403 let mut preview=Connection::open_in_memory()?;schema::initialize(&mut preview)?;
404 let tx=preview.transaction()?;
405 let mut mappings=Vec::new();
406 let mut preview_conflicts=Vec::new();
407 let mut preview_documents=Vec::new();
408 if let Err(e)=apply(&tx,&plan,&mut mappings,&mut preview_conflicts,&mut preview_documents){preview_conflicts.push(e.to_string());report.conflicts=preview_conflicts;return Ok(report)}
409 report.id_map=mappings;
410 {
411 let mut stmt=tx.prepare("SELECT kind,COUNT(*) FROM records GROUP BY kind")?;for row in stmt.query_map([],|r|Ok((r.get::<_,i64>(0)?,r.get::<_,i64>(1)?)))?{
412 let(code,count)=row?;let name=RecordKind::from_code(code).map(|k|k.as_str().to_string()).unwrap_or_else(||code.to_string());report.counts.insert(name,count as usize);}}
413 drop(tx);
414 if req.destination.exists()&&std::fs::read_dir(&req.destination)?.next().is_some(){
415 let database=req.destination.join("store.sqlite3");
416 if database.is_file(){
417 let conn=source_conn(&database)?;
418 let app:i64=conn.pragma_query_value(None,"application_id",|r|r.get(0))?;
419 if app==schema::APPLICATION_ID&&has_table(&conn,"import_runs")?{
420 let previous:Option<(String,String)>=conn.query_row("SELECT source_fingerprint,report_json FROM import_runs WHERE source_id=?1",[&req.source_id],|r|Ok((r.get(0)?,r.get(1)?))).optional()?;
421 if let Some((old,serialized))=previous{
422 if old==fingerprint{let mut previous:ImportReport=serde_json::from_str(&serialized)?;previous.already_imported=true;previous.applied=false;return Ok(previous)}
423 report.conflicts.push("Source changed since the previous import; use a new destination for a new snapshot.".into());return Ok(report)
424 }
425 }
426 }
427 report.conflicts.push("Destination must be a new or empty directory; existing user data is never overwritten.".into());return Ok(report)
428 }
429 if req.dry_run{report.conflicts=preview_conflicts;return Ok(report)}
430 let kb=KnowledgeBase::open(&req.destination)?;
431 let mut documents: Vec<crate::index::IndexDocument> = Vec::new();
432 kb.mutate(|tx|{
433 let count:i64=tx.query_row("SELECT COUNT(*) FROM records",[],|r|r.get(0))?;
434 if count!=0{return Err(Error::Conflict("destination became nonempty during import".into()))}
435 let mut mappings=Vec::new();report.conflicts.clear();apply(tx,&plan,&mut mappings,&mut report.conflicts,&mut documents)?;report.id_map=mappings;report.applied=true;
436 tx.execute("INSERT INTO import_runs(source_id,source_fingerprint,report_json) VALUES (?1,?2,?3)",params![req.source_id,fingerprint,serde_json::to_string(&report)?])?;
437 Ok(())
438 })?;
439 kb.index_documents(&documents)?;
441 match kb.update_index() { Ok(_) => {}, Err(error) => report.index_error = Some(error.to_string()) }
442 report.index_ready = report.index_error.is_none();
443 kb.write(|writer| Ok(writer.conn.execute("UPDATE import_runs SET report_json=?2 WHERE source_id=?1",params![req.source_id,serde_json::to_string(&report)?])?))?;
444 kb.close()?;Ok(report)
445}