]> git.giacomolli.uk - onechat/blob - whatsapp_import.go
Store shared private images by content hash
[onechat] / whatsapp_import.go
1 package main
2
3 import (
4 "archive/zip"
5 "bufio"
6 "context"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "io"
11 "maps"
12 "os"
13 "path/filepath"
14 "regexp"
15 "slices"
16 "strings"
17 "time"
18
19 "go.mau.fi/whatsmeow/store/sqlstore"
20 waLog "go.mau.fi/whatsmeow/util/log"
21 )
22
23 const (
24 // whatsappImportMark marks history in the first line of a batch.
25 whatsappImportMark = " (imported)"
26
27 // whatsappImportBytes limits one export's uncompressed chat text.
28 whatsappImportBytes = 64 << 20
29 )
30
31 var (
32 whatsappImportStart = regexp.MustCompile(`^\[(\d{2}/\d{2}/\d{4}, \d{2}:\d{2}:\d{2})\] ([^\r\n]+?): ?(.*)$`)
33 whatsappImportHeader = regexp.MustCompile(`^\[WhatsApp\] (.*), (?:group|private chat) ([^ ,]+), [A-Za-z]+ (\d{4}-\d{2}-\d{2})$`)
34 whatsappImportUnknown = regexp.MustCompile(`^\[WhatsApp\] (.*), [A-Za-z]+ (\d{4}-\d{2}-\d{2})$`)
35 )
36
37 // whatsappImport runs the offline import command. It holds the same token lock
38 // as the daemon and probe, even for a plan. The plan reads no credentials for a
39 // model and makes no network call. Only a completed rebuild replaces the agent
40 // directory. Unfinished subagents and pending target deliveries prevent either
41 // mode.
42 func whatsappImport(ctx context.Context, args []string, out io.Writer) (err error) {
43 options, err := whatsappImportParse(args)
44 if err != nil {
45 return fmt.Errorf("parse import arguments: %w", err)
46 }
47
48 dir, err := dataDir()
49 if err != nil {
50 return fmt.Errorf("find import data directory: %w", err)
51 }
52
53 tokens, err := lockTokens(dir)
54 if err != nil {
55 return fmt.Errorf("lock import tokens: %w", err)
56 }
57
58 defer func() {
59 closeErr := tokens.Close()
60 if closeErr != nil {
61 err = errors.Join(err, fmt.Errorf("close import token lock: %w", closeErr))
62 }
63 }()
64
65 root := filepath.Join(dir, "agent")
66 agent, err := whatsappImportAgent(root, options.agent)
67 if err != nil {
68 return fmt.Errorf("select import agent: %w", err)
69 }
70
71 err = whatsappImportIdle(root, agent)
72 if err != nil {
73 return fmt.Errorf("check import work: %w", err)
74 }
75
76 old, err := rebuildReadLog(ctx, agent.dir)
77 if err != nil {
78 return fmt.Errorf("read import log: %w", err)
79 }
80
81 var rootLog []DatedMessage
82
83 if agent.dir != root {
84 rootLog, err = rebuildReadLog(ctx, root)
85 if err != nil {
86 return fmt.Errorf("read root import history: %w", err)
87 }
88 }
89
90 names, owner, err := whatsappImportNames(ctx, dir)
91 if err != nil {
92 return fmt.Errorf("read import names: %w", err)
93 }
94
95 plan, err := whatsappImportPlan(ctx, options.paths, old, rootLog, names, owner, time.Local)
96 if err != nil {
97 return fmt.Errorf("plan import: %w", err)
98 }
99
100 plan.agent = agent.config.Name
101 err = plan.Write(out)
102 if err != nil {
103 return fmt.Errorf("write import plan: %w", err)
104 }
105
106 if options.planOnly || len(plan.added) == 0 {
107 return nil
108 }
109
110 // An existing agent is required so that a failed import never creates
111 // or changes the original agent's configuration.
112 _, err = os.Stat(filepath.Join(agent.dir, "config.json"))
113 if err != nil {
114 return fmt.Errorf("read agent config: %w", err)
115 }
116
117 config, err := loadConfig(dir)
118 if err != nil {
119 return fmt.Errorf("load import configuration: %w", err)
120 }
121
122 address, auth, err := modelEndpoint(config, dir)
123 if err != nil {
124 return fmt.Errorf("select import endpoint: %w", err)
125 }
126
127 usage, err := os.CreateTemp(dir, ".import-usage-*.jsonl")
128 if err != nil {
129 return fmt.Errorf("create import usage: %w", err)
130 }
131
132 usagePath := usage.Name()
133 defer func() {
134 closeErr := os.Remove(usagePath)
135 if closeErr != nil {
136 err = errors.Join(err, fmt.Errorf("remove import usage: %w", closeErr))
137 }
138 }()
139
140 err = usage.Close()
141 if err != nil {
142 return fmt.Errorf("close import usage: %w", err)
143 }
144
145 client := newModelClient(address, auth, agent.CacheKey(), usagePath)
146 backup, err := rebuild(ctx, agent, plan.added, client, newCompactor(client, agent.Prompt, config.Compactor, agent.Tools()), out)
147 if err != nil {
148 return fmt.Errorf("rebuild imported memory: %w", err)
149 }
150
151 _, err = fmt.Fprintf(out, "import complete: backup %s\n", backup)
152 if err != nil {
153 return fmt.Errorf("write import result: %w", err)
154 }
155
156 return nil
157 }
158
159 // WhatsAppImportPlan holds the new chat-day batches and the size of the merged
160 // log. Its estimate uses the measured cost of 1.3 calls per message, not a
161 // promise about a particular export or compaction retry.
162 type WhatsAppImportPlan struct {
163 agent string
164 cutoffs map[string]time.Time
165 added []DatedMessage
166 chats []string
167 cutoff time.Time
168 first time.Time
169 last time.Time
170 bytes int
171 messages int
172 skipped int
173 ownerKnown bool
174 }
175
176 // Write prints the plan without chat text or the owner's profile name.
177 func (p WhatsAppImportPlan) Write(out io.Writer) error {
178 cutoff := "none (no live WhatsApp batch)"
179 if !p.cutoff.IsZero() {
180 cutoff = p.cutoff.Format(time.RFC3339)
181 }
182
183 dateRange := "none"
184 if !p.first.IsZero() {
185 dateRange = p.first.Format(time.RFC3339) + " to " + p.last.Format(time.RFC3339)
186 }
187
188 owner := "known (mapped to You)"
189 if !p.ownerKnown {
190 owner = "unknown (sender names kept as written)"
191 }
192
193 _, err := fmt.Fprintf(out, "agent: %s\nchats: %d\n", p.agent, len(p.chats))
194 if err != nil {
195 return fmt.Errorf("write import plan: %w", err)
196 }
197
198 for _, chat := range p.chats {
199 line := " " + chat
200 cutoff, found := p.cutoffs[chat]
201 if found {
202 line += " (cutoff: " + cutoff.Format(time.RFC3339) + ")"
203 }
204
205 _, err = fmt.Fprintln(out, line)
206 if err != nil {
207 return fmt.Errorf("write import plan: %w", err)
208 }
209 }
210
211 estimate := 1.3 * float64(p.messages)
212 if len(p.added) == 0 {
213 estimate = 0
214 }
215
216 lines := []string{
217 fmt.Sprintf("batches: %d (%d chat-days skipped)", len(p.added), p.skipped),
218 "dates: " + dateRange,
219 "cutoff: " + cutoff,
220 "owner: " + owner,
221 fmt.Sprintf("bytes: %d (new batch text)", p.bytes),
222 fmt.Sprintf("messages: %d (merged log)", p.messages),
223 fmt.Sprintf("estimated calls: %.0f (about 1.3 per message, limit %d)", estimate, rebuildCallLimit),
224 }
225
226 _, err = io.WriteString(out, strings.Join(lines, "\n")+"\n")
227 if err != nil {
228 return fmt.Errorf("write import plan: %w", err)
229 }
230
231 return nil
232 }
233
234 // WhatsAppImportOptions selects the target, exports, and plan-only mode.
235 type WhatsAppImportOptions struct {
236 planOnly bool
237 agent string
238 paths []string
239 }
240
241 // whatsappImportParse accepts GNU-style flags, including --agent=NAME and --
242 // before paths that start with a dash.
243 func whatsappImportParse(args []string) (WhatsAppImportOptions, error) {
244 var options WhatsAppImportOptions
245
246 flags := true
247 for i := 0; i < len(args); i++ {
248 arg := args[i]
249 switch {
250 case flags && arg == "--":
251 flags = false
252 case flags && arg == "-n":
253 options.planOnly = true
254 case flags && (arg == "--agent" || strings.HasPrefix(arg, "--agent=")):
255 _, name, found := strings.Cut(arg, "=")
256 if !found {
257 i++
258 if i == len(args) {
259 return options, errors.New("--agent needs a child name")
260 }
261
262 name = args[i]
263 }
264
265 if name == "" {
266 return options, errors.New("--agent needs a child name")
267 }
268
269 options.agent = name
270 case flags && strings.HasPrefix(arg, "-"):
271 return options, fmt.Errorf("unknown import flag %q", arg)
272 default:
273 options.paths = append(options.paths, arg)
274 }
275 }
276
277 if len(options.paths) == 0 {
278 return options, errors.New("usage: onechat import [-n] [--agent NAME] ZIP [ZIP...]")
279 }
280
281 return options, nil
282 }
283
284 // whatsappImportAgent selects the root or an existing direct child without
285 // creating a directory or config. A root without a config can have a plan,
286 // using the default name.
287 func whatsappImportAgent(root, name string) (*Agent, error) {
288 _, err := os.Stat(filepath.Join(root, "config.json"))
289 if errors.Is(err, os.ErrNotExist) && name == "" {
290 var config AgentConfig
291
292 config, err = defaultAgentConfig()
293 if err != nil {
294 return nil, fmt.Errorf("read default import agent: %w", err)
295 }
296
297 return &Agent{
298 dir: root,
299 config: config,
300 }, nil
301 }
302
303 if err != nil {
304 return nil, fmt.Errorf("read root agent config: %w", err)
305 }
306
307 agent, err := openAgent(root)
308 if err != nil {
309 return nil, fmt.Errorf("open root agent: %w", err)
310 }
311
312 if name == "" {
313 return agent, nil
314 }
315
316 if agent.config.Parent != "" {
317 return nil, errors.New("import root cannot have a parent")
318 }
319
320 children, err := agent.Children()
321 if err != nil {
322 return nil, fmt.Errorf("read import children: %w", err)
323 }
324
325 for _, child := range children {
326 if child.config.Name == name {
327 return child, nil
328 }
329 }
330
331 return nil, fmt.Errorf("child %q does not exist", name)
332 }
333
334 // whatsappImportIdle refuses unfinished subagents and pending deliveries to the
335 // target. Rebuilding renumbers its log, so saved delivery IDs must not refer to
336 // it until their turns finish. The caller holds the token lock.
337 func whatsappImportIdle(root string, agent *Agent) error {
338 err := rebuildIdle(agent.dir)
339 if err != nil {
340 return fmt.Errorf("check target work: %w", err)
341 }
342
343 data, err := os.ReadFile(filepath.Join(root, "messages.json"))
344 if errors.Is(err, os.ErrNotExist) {
345 return nil
346 }
347
348 if err != nil {
349 return fmt.Errorf("read agent deliveries: %w", err)
350 }
351
352 var queue DaemonMessageFile
353
354 err = json.Unmarshal(data, &queue)
355 if err != nil {
356 return fmt.Errorf("decode agent deliveries: %w", err)
357 }
358
359 for _, message := range queue.Pending {
360 if message.Target == agent.config.Name {
361 return fmt.Errorf("import refused: agent %q has pending deliveries", agent.config.Name)
362 }
363 }
364
365 return nil
366 }
367
368 // whatsappImportNames reads only local session data. Group names that the
369 // library keeps only in memory can also come from logged or pending batches.
370 // Ambiguous names have no ID. A missing session keeps sender names unchanged.
371 func whatsappImportNames(ctx context.Context, dir string) (names map[string]WhatsAppChat, owner string, err error) {
372 names = make(map[string]WhatsAppChat)
373 path := filepath.Join(dir, whatsappDirName, whatsappSessionFileName)
374 _, err = os.Stat(path)
375 if errors.Is(err, os.ErrNotExist) {
376 return names, "", nil
377 }
378
379 if err != nil {
380 return nil, "", fmt.Errorf("read import session: %w", err)
381 }
382
383 address, err := whatsappSessionAddress(path)
384 if err != nil {
385 return nil, "", fmt.Errorf("find import session address: %w", err)
386 }
387
388 container, err := sqlstore.New(ctx, "sqlite", address, waLog.Noop)
389 if err != nil {
390 return nil, "", fmt.Errorf("open import session: %w", err)
391 }
392
393 defer func() {
394 closeErr := container.Close()
395 if closeErr != nil {
396 err = errors.Join(err, fmt.Errorf("close import session: %w", closeErr))
397 }
398 }()
399
400 device, err := container.GetFirstDevice(ctx)
401 if err != nil {
402 return nil, "", fmt.Errorf("read import device: %w", err)
403 }
404
405 if device.ID == nil {
406 return names, "", nil
407 }
408
409 owner = device.PushName
410 own := whatsappOwnIDs(device.ID.User, device.LID.User)
411 contacts, err := device.Contacts.GetAllContacts(ctx)
412 if err != nil {
413 return nil, "", fmt.Errorf("read import contacts: %w", err)
414 }
415
416 for jid, contact := range contacts {
417 if slices.Contains(own, jid.User) {
418 continue
419 }
420
421 for _, name := range []string{contact.FirstName, contact.FullName, contact.PushName, contact.BusinessName} {
422 if name != "" {
423 whatsappImportName(names, WhatsAppChat{
424 ID: jid.ToNonAD().String(),
425 Name: name,
426 Group: jid.Server == "g.us",
427 })
428 }
429 }
430 }
431
432 data, err := os.ReadFile(filepath.Join(dir, whatsappDirName, whatsappStateFileName))
433 if errors.Is(err, os.ErrNotExist) {
434 return names, owner, nil
435 }
436
437 if err != nil {
438 return nil, "", fmt.Errorf("read pending import names: %w", err)
439 }
440
441 var state WhatsAppStateFile
442
443 err = json.Unmarshal(data, &state)
444 if err != nil {
445 return nil, "", fmt.Errorf("decode pending import names: %w", err)
446 }
447
448 for _, batch := range state.Batches {
449 whatsappImportName(names, batch.Chat)
450 }
451
452 return names, owner, nil
453 }
454
455 // whatsappImportName adds an exact match unless distinct IDs share its name.
456 func whatsappImportName(names map[string]WhatsAppChat, chat WhatsAppChat) {
457 old, found := names[chat.Name]
458 if found && old.ID != chat.ID {
459 chat.ID = ""
460 }
461
462 names[chat.Name] = chat
463 }
464
465 // whatsappImportPlan filters each chat at its earliest live message across the
466 // target and root logs. Only imported days in the target are skipped. Each
467 // remaining chat-day becomes a work message dated at its last exported message.
468 // Overlapping input chat-days are refused.
469 func whatsappImportPlan(ctx context.Context, paths []string, old, rootLog []DatedMessage, names map[string]WhatsAppChat, owner string, location *time.Location) (WhatsAppImportPlan, error) {
470 plan := WhatsAppImportPlan{
471 ownerKnown: owner != "",
472 cutoffs: make(map[string]time.Time),
473 }
474
475 names = maps.Clone(names)
476 if names == nil {
477 names = make(map[string]WhatsAppChat)
478 }
479
480 imported, err := whatsappImportHistory(old, rootLog, names, plan.cutoffs, location)
481 if err != nil {
482 return plan, fmt.Errorf("read import history: %w", err)
483 }
484
485 seen := make(map[string]bool)
486 for _, path := range paths {
487 if ctx.Err() != nil {
488 return plan, ctx.Err()
489 }
490
491 chat, messages, err := whatsappImportRead(path, owner, location)
492 if err != nil {
493 return plan, fmt.Errorf("read import export: %w", err)
494 }
495
496 known, found := names[chat.Name]
497 if found {
498 chat = known
499 }
500
501 if !slices.Contains(plan.chats, chat.Name) {
502 plan.chats = append(plan.chats, chat.Name)
503 }
504
505 cutoff := plan.cutoffs[chat.Name]
506 if !cutoff.IsZero() && (plan.cutoff.IsZero() || cutoff.Before(plan.cutoff)) {
507 plan.cutoff = cutoff
508 }
509
510 days := make(map[string][]WhatsAppMessage)
511 for _, message := range messages {
512 if !cutoff.IsZero() && !message.Time.Before(cutoff) {
513 continue
514 }
515
516 day := message.Time.In(location).Format(time.DateOnly)
517 days[day] = append(days[day], message)
518 }
519
520 for _, day := range slices.Sorted(maps.Keys(days)) {
521 key := chat.Name + "\n" + day
522 if imported[key] {
523 plan.skipped++
524 continue
525 }
526
527 if seen[key] {
528 return plan, errors.New("import inputs overlap: give only one export for each chat and day")
529 }
530
531 seen[key] = true
532 batch := WhatsAppBatch{
533 Chat: chat,
534 Messages: days[day],
535 }
536
537 text := batch.Text(location)
538 first, body, _ := strings.Cut(text, "\n")
539 if chat.ID == "" {
540 first = "[WhatsApp] " + chat.Name + ", " + batch.Messages[0].Time.In(location).Format(whatsappBatchDayLayout)
541 }
542
543 text = first + whatsappImportMark + "\n" + body
544 plan.added = append(plan.added, DatedMessage{
545 Kind: "work",
546 Text: text,
547 Date: batch.Messages[len(batch.Messages)-1].Time.Format(time.RFC3339Nano),
548 })
549
550 plan.bytes += len(text)
551 firstDate, lastDate := batch.Messages[0].Time, batch.Messages[len(batch.Messages)-1].Time
552 if plan.first.IsZero() || firstDate.Before(plan.first) {
553 plan.first = firstDate
554 }
555
556 if plan.last.IsZero() || lastDate.After(plan.last) {
557 plan.last = lastDate
558 }
559 }
560 }
561
562 plan.added, err = rebuildMerge(nil, plan.added)
563 if err != nil {
564 return plan, fmt.Errorf("order import batches: %w", err)
565 }
566
567 plan.messages = len(old) + len(plan.added)
568 slices.Sort(plan.chats)
569
570 return plan, nil
571 }
572
573 // whatsappImportHistory reads batch names and live cutoffs from both logs.
574 // Imported chat-days belong only to the target log, which is first.
575 func whatsappImportHistory(target, root []DatedMessage, names map[string]WhatsAppChat, cutoffs map[string]time.Time, location *time.Location) (map[string]bool, error) {
576 imported := make(map[string]bool)
577 for source, entries := range [][]DatedMessage{target, root} {
578 for _, entry := range entries {
579 if entry.Kind != "work" || !strings.HasPrefix(entry.Text, "[WhatsApp] ") {
580 continue
581 }
582
583 first, _, _ := strings.Cut(entry.Text, "\n")
584 marked := strings.HasSuffix(first, whatsappImportMark)
585 first = strings.TrimSuffix(first, whatsappImportMark)
586 match := whatsappImportHeader.FindStringSubmatch(first)
587 name, day := "", ""
588 if len(match) > 0 {
589 name, day = match[1], match[3]
590 whatsappImportName(names, WhatsAppChat{
591 ID: match[2],
592 Name: name,
593 Group: strings.Contains(first, ", group "+match[2]+","),
594 })
595 } else if marked {
596 match = whatsappImportUnknown.FindStringSubmatch(first)
597 if len(match) > 0 {
598 name, day = match[1], match[2]
599 }
600 }
601
602 if name == "" {
603 return nil, errors.New("invalid WhatsApp batch header")
604 }
605
606 if marked {
607 if source == 0 {
608 imported[name+"\n"+day] = true
609 }
610
611 continue
612 }
613
614 _, body, found := strings.Cut(entry.Text, "\n")
615 if !found || len(body) < len("15:04 ") {
616 return nil, errors.New("missing first WhatsApp message time")
617 }
618
619 date, err := time.ParseInLocation("2006-01-02 15:04", day+" "+body[:5], location)
620 if err != nil {
621 return nil, fmt.Errorf("parse first live message time: %w", err)
622 }
623
624 if cutoffs[name].IsZero() || date.Before(cutoffs[name]) {
625 cutoffs[name] = date
626 }
627 }
628 }
629
630 return imported, nil
631 }
632
633 // whatsappImportRead reads the single _chat.txt in an export without extracting
634 // archive paths. Non-start lines continue the preceding message, and all
635 // invisible left-to-right marks are removed. Dates use location. Media and
636 // system messages remain text, including the written sender.
637 func whatsappImportRead(path, owner string, location *time.Location) (chat WhatsAppChat, messages []WhatsAppMessage, err error) {
638 chat.Name = strings.TrimPrefix(strings.TrimSuffix(filepath.Base(path), filepath.Ext(path)), "WhatsApp Chat - ")
639 if chat.Name == "" || strings.ContainsAny(chat.Name, "\r\n") {
640 return chat, nil, errors.New("read export: chat name must be one nonempty line")
641 }
642
643 archive, err := zip.OpenReader(path)
644 if err != nil {
645 return chat, nil, fmt.Errorf("open export: %w", err)
646 }
647
648 defer func() {
649 closeErr := archive.Close()
650 if closeErr != nil {
651 err = errors.Join(err, fmt.Errorf("close export: %w", closeErr))
652 }
653 }()
654
655 var file *zip.File
656
657 for _, entry := range archive.File {
658 if filepath.Base(entry.Name) == "_chat.txt" && !entry.FileInfo().IsDir() {
659 if file != nil {
660 return chat, nil, errors.New("read export: more than one _chat.txt")
661 }
662
663 file = entry
664 }
665 }
666
667 if file == nil {
668 return chat, nil, errors.New("read export: _chat.txt is missing")
669 }
670
671 if file.UncompressedSize64 > whatsappImportBytes {
672 return chat, nil, errors.New("read export: chat text exceeds 64 MiB")
673 }
674
675 reader, err := file.Open()
676 if err != nil {
677 return chat, nil, fmt.Errorf("open chat text: %w", err)
678 }
679
680 defer func() {
681 closeErr := reader.Close()
682 if closeErr != nil {
683 err = errors.Join(err, fmt.Errorf("close export chat text: %w", closeErr))
684 }
685 }()
686
687 scanner := bufio.NewScanner(io.LimitReader(reader, whatsappImportBytes+1))
688 scanner.Buffer(make([]byte, 64*1024), buildLineBytes)
689
690 for scanner.Scan() {
691 line := strings.ReplaceAll(scanner.Text(), "\u200e", "")
692 match := whatsappImportStart.FindStringSubmatch(line)
693 if match == nil {
694 if len(messages) == 0 {
695 if line == "" {
696 continue
697 }
698
699 return chat, nil, errors.New("read export: text before the first dated message")
700 }
701
702 messages[len(messages)-1].Text += "\n" + line
703
704 continue
705 }
706
707 var date time.Time
708
709 date, err = time.ParseInLocation("02/01/2006, 15:04:05", match[1], location)
710 if err != nil {
711 return chat, nil, fmt.Errorf("parse export local date: %w", err)
712 }
713
714 messages = append(messages, WhatsAppMessage{
715 Sender: match[2],
716 Text: match[3],
717 Time: date,
718 Mine: owner != "" && match[2] == owner && match[2] != chat.Name,
719 })
720
721 if match[2] == chat.Name {
722 chat.Group = true
723 }
724 }
725
726 if scanner.Err() != nil {
727 return chat, nil, fmt.Errorf("read chat text: %w", scanner.Err())
728 }
729
730 if len(messages) == 0 {
731 return chat, nil, errors.New("read export: no dated messages")
732 }
733
734 slices.SortStableFunc(messages, func(a, b WhatsAppMessage) int {
735 return a.Time.Compare(b.Time)
736 })
737
738 return chat, messages, nil
739 }