(userID string, zr *zip.Reader)
| 489 | } |
| 490 | |
| 491 | func (r *run) importTweetsFromZip(userID string, zr *zip.Reader) error { |
| 492 | log.Printf("twitter: processing zip file with %d files", len(zr.File)) |
| 493 | |
| 494 | tweetsNode, err := r.getTopLevelNode("tweets") |
| 495 | if err != nil { |
| 496 | return err |
| 497 | } |
| 498 | |
| 499 | var ( |
| 500 | gate = syncutil.NewGate(tweetsAtOnce) |
| 501 | grp syncutil.Group |
| 502 | ) |
| 503 | total := 0 |
| 504 | for _, zf := range zr.File { |
| 505 | if !(strings.HasPrefix(zf.Name, "data/js/tweets/2") && strings.HasSuffix(zf.Name, ".js")) { |
| 506 | continue |
| 507 | } |
| 508 | tweets, err := tweetsFromZipFile(zf) |
| 509 | if err != nil { |
| 510 | return fmt.Errorf("error reading tweets from %s: %v", zf.Name, err) |
| 511 | } |
| 512 | |
| 513 | for i := range tweets { |
| 514 | total++ |
| 515 | tweet := tweets[i] |
| 516 | gate.Start() |
| 517 | grp.Go(func() error { |
| 518 | defer gate.Done() |
| 519 | _, err := r.importTweet(tweetsNode, tweet, false) |
| 520 | return err |
| 521 | }) |
| 522 | } |
| 523 | } |
| 524 | err = grp.Err() |
| 525 | log.Printf("zip import of tweets: %d total, err = %v", total, err) |
| 526 | return err |
| 527 | } |
| 528 | |
| 529 | func timeParseFirstFormat(timeStr string, format ...string) (t time.Time, err error) { |
| 530 | if len(format) == 0 { |
no test coverage detected