switch disqus importer to streaming model

2017-12-24 12:18:03 -06:00
parent 3fa4ab418d
commit c5d8b90a1d
5 changed files with 92 additions and 25 deletions
@@ -7,7 +7,7 @@ RUN cd app && go test -v $(go list -e ./... | grep -v vendor)

 RUN gometalinter --disable-all --deadline=300s --vendor --enable=vet --enable=vetshadow --enable=golint \
    --enable=staticcheck --enable=ineffassign --enable=goconst --enable=errcheck --enable=unconvert \
-    --enable=deadcode  --enable=gosimple --enable=gas --exclude=test ./...
+    --enable=deadcode  --enable=gosimple --enable=gas --exclude=test --exclude=mock ./...

 #RUN /script/checkvendor.sh
 RUN mkdir -p target && /script/coverage.sh
@@ -7,6 +7,10 @@ import (
 	"log"
 	"time"

+	"sync"
+
+	"strings"
+
 	"github.com/pkg/errors"
 	"github.com/umputun/remark/app/store"
 )
@@ -14,6 +18,9 @@ import (
 // Disqus implements Importer from disqus xml
 type Disqus struct {
 	DataStore store.Interface
+
+	ch   chan store.Comment
+	once sync.Once
 }

 type disqusXML struct {
@@ -28,7 +35,7 @@ type disqusThread struct {
 	Title       string    `xml:"title"`
 	Message     string    `xml:"message"`
 	CreateAt    time.Time `xml:"createdAt"`
-	AuthorNmae  string    `xml:"author>name"`
+	AuthorName  string    `xml:"author>name"`
 	AuthorEmail string    `xml:"author>email"`
 	Anonymous   bool      `xml:"author>isAnonymous"`
 	IP          string    `xml:"ipAddress"`
@@ -54,7 +61,8 @@ type uid struct {
 }

 // Import from disqus and save to store
-func (d Disqus) Import(r io.Reader) (err error) {
+func (d *Disqus) Import(r io.Reader, siteID string) (err error) {
+
 	data, err := ioutil.ReadAll(r)
 	if err != nil {
 		return errors.Wrap(err, "failed to read data")
@@ -64,8 +72,9 @@ func (d Disqus) Import(r io.Reader) (err error) {
 		return errors.Wrap(err, "can't unmarshal disqus xml")
 	}

+	commentsCh := d.convert(r, siteID)
 	failed := 0
-	for _, c := range d.convert(dxml) {
+	for c := range commentsCh {
 		if _, err = d.DataStore.Create(c); err != nil {
 			failed++
 		}
@@ -74,25 +83,56 @@ func (d Disqus) Import(r io.Reader) (err error) {
 	if failed > 0 {
 		return errors.Errorf("failed to save %d comments", failed)
 	}
+
 	return nil
 }

-func (d Disqus) convert(dxml disqusXML) (comments []store.Comment) {
+func (d *Disqus) convert(r io.Reader, siteID string) (ch chan store.Comment) {
+
 	postsMap := map[string]string{} // tid:url
-	log.Printf("[DEBUG] convert %d posts, %d comments", len(dxml.Threads), len(dxml.Comments))
-	for _, thread := range dxml.Threads {
-		postsMap[thread.UID] = thread.Link
-	}
-	for _, comment := range dxml.Comments {
-		c := store.Comment{
-			ID:        comment.ID,
-			Locator:   store.Locator{URL: postsMap[comment.Tid.Val]},
-			User:      store.User{ID: comment.AuthorUserName, Name: comment.AuthorName, IP: comment.IP},
-			Text:      comment.Message,
-			Timestamp: comment.CreatedAt,
-			ParentID:  comment.Pid.Val,
+	decoder := xml.NewDecoder(r)
+	commentsCh := make(chan store.Comment)
+
+	go func() {
+		commentsCount := 0
+		for {
+			t, err := decoder.Token()
+			if t == nil || err != nil {
+				break
+			}
+
+			switch se := t.(type) {
+			case xml.StartElement:
+				if se.Name.Local == "thread" {
+					thread := disqusThread{}
+					decoder.DecodeElement(&thread, &se)
+					postsMap[thread.UID] = thread.Link
+				}
+				if se.Name.Local == "post" {
+					comment := disqusComment{}
+					decoder.DecodeElement(&comment, &se)
+					c := store.Comment{
+						ID:        comment.ID,
+						Locator:   store.Locator{URL: postsMap[comment.Tid.Val], SiteID: siteID},
+						User:      store.User{ID: comment.AuthorUserName, Name: comment.AuthorName, IP: comment.IP},
+						Text:      d.cleanText(comment.Message),
+						Timestamp: comment.CreatedAt,
+						ParentID:  comment.Pid.Val,
+					}
+					commentsCh <- c
+					commentsCount++
+				}
+			}
 		}
-		comments = append(comments, c)
-	}
-	return comments
+		close(commentsCh)
+		log.Printf("[DEBUG] converted %d posts, %d comments", len(postsMap), commentsCount)
+	}()
+
+	return commentsCh
+}
+
+func (d *Disqus) cleanText(text string) string {
+	text = strings.Replace(text, "\n", "", -1)
+	text = strings.Replace(text, "\t", "", -1)
+	return text
 }
@@ -4,13 +4,38 @@ import (
 	"strings"
 	"testing"

+	"time"
+
 	"github.com/stretchr/testify/assert"
+	"github.com/umputun/remark/app/store"
 )

-func TestA(t *testing.T) {
+func TestDisqus_Convert(t *testing.T) {
 	d := Disqus{}
-	err := d.Import(strings.NewReader(xmlTest))
-	assert.Nil(t, err)
+	ch := d.convert(strings.NewReader(xmlTest), "test")
+
+	res := []store.Comment{}
+	for comment := range ch {
+		res = append(res, comment)
+		t.Logf("%+v", comment)
+	}
+	assert.Equal(t, 3, len(res), "3 comments total")
+
+	exp0 := store.Comment{
+		ID: "3565798471341011339",
+		Locator: store.Locator{
+			SiteID: "test",
+			URL:    "https://radio-t.com/p/2011/03/05/podcast-229/",
+		},
+		Text: "<p>The quick brown fox jumps over the lazy dog.</p>",
+		User: store.User{
+			Name: "Alexander Puzatykh",
+			ID:   "facebook-1787732238",
+			IP:   "178.234.205.125",
+		},
+	}
+	exp0.Timestamp, _ = time.Parse("2006-01-02T15:04:05Z", "2011-08-31T15:16:29Z")
+	assert.Equal(t, exp0, res[0])
 }

 var xmlTest = `
@@ -57,7 +82,7 @@ var xmlTest = `
 	<post dsq:id="299619020">
 		<id>3565798471341011339</id>
 		<message>
-			<![CDATA[<p>Недавно посмотрел фильм про Джобса и Aple. Тот он еще был перец. <a href="http://alex-bestbusiness.com/" rel="nofollow noopener" title="http://alex-bestbusiness.com/">http://alex-bestbusiness.com/</a></p>]]>
+			<![CDATA[<p>The quick brown fox jumps over the lazy dog.</p>]]>
 		</message>
 		<createdAt>2011-08-31T15:16:29Z</createdAt>
 		<isDeleted>false</isDeleted>
@@ -4,5 +4,5 @@ import "io"

 // Importer defines interface to convert posts from external sources
 type Importer interface {
-	Import(r io.Reader) error
+	Import(r io.Reader, siteID string) error
 }
@@ -1,5 +1,7 @@
 package store

+//go:generate sh -c "mockery -inpkg -name Interface -print > file.tmp && mv file.tmp store_mock.go"
+
 import (
 	"crypto/rand"
 	"crypto/sha1"