fix-migration-replyto

migration/tables/content_items.py

@@ -9,7 +9,9 @@ from orm.reaction import Reaction, ReactionKind
 from orm.shout import Shout, ShoutTopic, ShoutReactionsFollower
 from orm.user import User
 from orm.topic import TopicFollower, Topic
+# from services.stat.reacted import ReactedStorage
 from services.stat.viewed import ViewedStorage
+import re
 
 OLD_DATE = "2016-03-05 22:22:00.350000"
 ts = datetime.now(tz=timezone.utc)
@@ -21,6 +23,8 @@ type2layout = {
     "Image": "image",
 }
 
+anondict = {"slug": "anonymous", "id": 1, "name": "Аноним"}
+
 
 def get_shout_slug(entry):
     slug = entry.get("slug", "")
@@ -29,6 +33,7 @@ def get_shout_slug(entry):
             slug = friend.get("slug", "")
             if slug:
                 break
+    slug = re.sub('[^0-9a-zA-Z]+', '-', slug)
     return slug
 
 
@@ -39,13 +44,8 @@ def create_author_from_app(app):
             user = session.query(User).where(User.email == app['email']).first()
             if not user:
                 name = app.get('name')
-                slug = (
-                    translit(name, "ru", reversed=True)
-                        .replace(" ", "-")
-                        .replace("'", "")
-                        .replace(".", "-")
-                        .lower()
-                )
+                slug = translit(name, "ru", reversed=True).lower()
+                slug = re.sub('[^0-9a-zA-Z]+', '-', slug)
                 # check if nameslug is used
                 user = session.query(User).where(User.slug == slug).first()
                 # get slug from email
@@ -74,122 +74,99 @@ def create_author_from_app(app):
                     session.commit()
             userdata = user.dict()
         if not userdata:
-            userdata = User.default_user.dict()
+            userdata = User.default_user.dict()  # anonymous
     except Exception as e:
         print(app)
         raise e
     return userdata
 
 
-async def create_shout(shout_dict, userslug):
+async def create_shout(shout_dict, user):
     s = Shout.create(**shout_dict)
     with local_session() as session:
-        follower = session.query(User).where(User.slug == userslug).one()
-
-        srf = session.query(
-            ShoutReactionsFollower
-        ).join(
-            User
-        ).where(
+        srf = session.query(ShoutReactionsFollower).where(
             ShoutReactionsFollower.shout == s.id
         ).filter(
-            User.slug == userslug
+            ShoutReactionsFollower.follower == user.id
         ).first()
         if not srf:
-            srf = ShoutReactionsFollower.create(shout=s.id, follower=follower.id, auto=True)
+            srf = ShoutReactionsFollower.create(shout=s.id, follower=user.id, auto=True)
             session.add(srf)
         session.commit()
+    return s
+
+
+def get_userdata(entry, storage):
+    user_oid = entry.get("createdBy", "")
+    userdata = None
+    app = entry.get("application")
+    if app:
+        userdata = create_author_from_app(app) or anondict
+    else:
+        userdata = storage["users"]["by_oid"].get(user_oid) or anondict
+    slug = userdata.get("slug")
+    slug = re.sub('[^0-9a-zA-Z]+', '-', slug)
+    userdata["slug"] = slug
+    return userdata, user_oid
 
 
 async def migrate(entry, storage):
-    # init, set title and layout
+    userdata, user_oid = get_userdata(entry, storage)
+    user = await get_user(userdata, storage, user_oid)
     r = {
         "layout": type2layout[entry["type"]],
         "title": entry["title"],
-        "authors": [],
-        "topics": set([])
+        "authors": [userdata["slug"], ],
+        "slug": get_shout_slug(entry),
+        "cover": (
+            "https://assets.discours.io/unsafe/1600x/" +
+            entry["thumborId"] if entry.get("thumborId") else entry.get("image", {}).get("url")
+        ),
+        "visibility": "public" if entry.get("published") else "authors",
+        "publishedAt": date_parse(entry.get("publishedAt")) if entry.get("published") else None,
+        "deletedAt": date_parse(entry.get("deletedAt")) if entry.get("deletedAt") else None,
+        "createdAt": date_parse(entry.get("createdAt", OLD_DATE)),
+        "updatedAt": date_parse(entry["updatedAt"]) if "updatedAt" in entry else ts,
+        "topics": await add_topics_follower(entry, storage, user),
+        "body": extract_html(entry)
     }
 
-    # author
-    users_by_oid = storage["users"]["by_oid"]
-    user_oid = entry.get("createdBy", "")
-    userdata = users_by_oid.get(user_oid)
-    user = None
-    if not userdata:
-        app = entry.get("application")
-        if app:
-            userdata = create_author_from_app(app)
-    if userdata:
-        userslug = userdata.get('slug')
-    else:
-        userslug = "anonymous"  # bad old id slug was found
-    r["authors"] = [userslug, ]
+    # main topic patch
+    r['mainTopic'] = r['topics'][0]
 
-    # slug
-    slug = get_shout_slug(entry)
-    if slug:
-        r["slug"] = slug
-    else:
-        raise Exception
-
-    # cover
-    c = ""
-    if entry.get("thumborId"):
-        c = "https://assets.discours.io/unsafe/1600x/" + entry["thumborId"]
-    else:
-        c = entry.get("image", {}).get("url")
-        if not c or "cloudinary" in c:
-            c = ""
-    r["cover"] = c
-
-    # timestamps
-    r["createdAt"] = date_parse(entry.get("createdAt", OLD_DATE))
-    r["updatedAt"] = date_parse(entry["updatedAt"]) if "updatedAt" in entry else ts
-
-    # visibility
+    # published author auto-confirm
     if entry.get("published"):
-        r["publishedAt"] = date_parse(entry.get("publishedAt", OLD_DATE))
-        r["visibility"] = "public"
         with local_session() as session:
             # update user.emailConfirmed if published
-            author = session.query(User).where(User.slug == userslug).first()
+            author = session.query(User).where(User.slug == userdata["slug"]).first()
             author.emailConfirmed = True
             session.add(author)
             session.commit()
-    else:
-        r["visibility"] = "authors"
 
-    if "deletedAt" in entry:
-        r["deletedAt"] = date_parse(entry["deletedAt"])
-
-    # topics
-    r['topics'] = await add_topics_follower(entry, storage, userslug)
-    r['mainTopic'] = r['topics'][0]
-
-    entry["topics"] = r["topics"]
-    entry["cover"] = r["cover"]
-
-    # body
-    r["body"] = extract_html(entry)
+    # media
     media = extract_media(entry)
-    if media:
-        r["media"] = json.dumps(media, ensure_ascii=True)
+    r["media"] = json.dumps(media, ensure_ascii=True) if media else None
 
+    # ----------------------------------- copy
     shout_dict = r.copy()
 
     # user
-    user = await get_user(userslug, userdata, storage, user_oid)
     shout_dict["authors"] = [user, ]
     del shout_dict["topics"]
     try:
         # save shout to db
-        await create_shout(shout_dict, userslug)
+        shout_dict["oid"] = entry.get("_id", "")
+        shout = await create_shout(shout_dict, user)
     except IntegrityError as e:
-        print(e)
-        await resolve_create_shout(shout_dict, userslug)
+        print('[migration] create_shout integrity error', e)
+        shout = await resolve_create_shout(shout_dict, userdata["slug"])
     except Exception as e:
         raise Exception(e)
 
+    # udpate data
+    shout_dict = shout.dict()
+    shout_dict["authors"] = [user.dict(), ]
+
     # shout topics aftermath
     shout_dict["topics"] = await topics_aftermath(r, storage)
 
@@ -200,13 +177,12 @@ async def migrate(entry, storage):
     await ViewedStorage.increment(shout_dict["slug"], amount=entry.get("views", 1))
     # del shout_dict['ratings']
 
-    shout_dict["oid"] = entry.get("_id", "")
     storage["shouts"]["by_oid"][entry["_id"]] = shout_dict
-    storage["shouts"]["by_slug"][slug] = shout_dict
+    storage["shouts"]["by_slug"][shout_dict["slug"]] = shout_dict
     return shout_dict
 
 
-async def add_topics_follower(entry, storage, userslug):
+async def add_topics_follower(entry, storage, user):
     topics = set([])
     category = entry.get("category")
     topics_by_oid = storage["topics"]["by_oid"]
@@ -218,29 +194,26 @@ async def add_topics_follower(entry, storage, userslug):
     ttt = list(topics)
     # add author as TopicFollower
     with local_session() as session:
-        for tpc in topics:
+        for tpcslug in topics:
             try:
-                topic = session.query(Topic).where(Topic.slug == tpc).one()
-                follower = session.query(User).where(User.slug == userslug).one()
-
+                tpc = session.query(Topic).where(Topic.slug == tpcslug).first()
                 tf = session.query(
                     TopicFollower
                 ).where(
-                    TopicFollower.follower == follower.id
+                    TopicFollower.follower == user.id
                 ).filter(
-                    TopicFollower.topic == topic.id
+                    TopicFollower.topic == tpc.id
                 ).first()
-
                 if not tf:
                     tf = TopicFollower.create(
-                        topic=topic.id,
-                        follower=follower.id,
+                        topic=tpc.id,
+                        follower=user.id,
                         auto=True
                     )
                     session.add(tf)
                     session.commit()
             except IntegrityError:
-                print('[migration.shout] hidden by topic ' + tpc)
+                print('[migration.shout] hidden by topic ' + tpc.slug)
     # main topic
     maintopic = storage["replacements"].get(topics_by_oid.get(category, {}).get("slug"))
     if maintopic in ttt:
@@ -249,23 +222,28 @@ async def add_topics_follower(entry, storage, userslug):
     return ttt
 
 
-async def get_user(userslug, userdata, storage, oid):
+async def get_user(userdata, storage, oid):
     user = None
     with local_session() as session:
-        if not user and userslug:
-            user = session.query(User).filter(User.slug == userslug).first()
-        if not user and userdata:
+        uid = userdata.get("id")
+        if uid:
+            user = session.query(User).filter(User.id == uid).first()
+        elif userdata:
             try:
-                userdata["slug"] = userdata["slug"].lower().strip().replace(" ", "-")
+                slug = userdata["slug"].lower().strip()
+                slug = re.sub('[^0-9a-zA-Z]+', '-', slug)
+                userdata["slug"] = slug
                 user = User.create(**userdata)
                 session.add(user)
                 session.commit()
             except IntegrityError:
-                print("[migration] user error: " + userdata)
-            userdata["id"] = user.id
-            userdata["createdAt"] = user.createdAt
-            storage["users"]["by_slug"][userdata["slug"]] = userdata
-            storage["users"]["by_oid"][oid] = userdata
+                print("[migration] user creating with slug %s" % userdata["slug"])
+                print("[migration] from userdata: %r" % userdata)
+                raise Exception("[migration] cannot create user in content_items.get_user()")
+        userdata["id"] = user.id
+        userdata["createdAt"] = user.createdAt
+        storage["users"]["by_slug"][userdata["slug"]] = userdata
+        storage["users"]["by_oid"][oid] = userdata
     if not user:
         raise Exception("could not get a user")
     return user
@@ -303,6 +281,7 @@ async def resolve_create_shout(shout_dict, userslug):
             print("[migration] something went wrong with shout: \n%r" % shout_dict)
             raise Exception("")
         session.commit()
+        return s
 
 
 async def topics_aftermath(entry, storage):
@@ -318,22 +297,22 @@ async def topics_aftermath(entry, storage):
 
                 shout_topic_old = (
                     session.query(ShoutTopic)
-                        .join(Shout)
-                        .join(Topic)
-                        .filter(Shout.slug == entry["slug"])
-                        .filter(Topic.slug == oldslug)
-                        .first()
+                    .join(Shout)
+                    .join(Topic)
+                    .filter(Shout.slug == entry["slug"])
+                    .filter(Topic.slug == oldslug)
+                    .first()
                 )
                 if shout_topic_old:
                     shout_topic_old.update({"topic": new_topic.id})
                 else:
                     shout_topic_new = (
                         session.query(ShoutTopic)
-                            .join(Shout)
-                            .join(Topic)
-                            .filter(Shout.slug == entry["slug"])
-                            .filter(Topic.slug == newslug)
-                            .first()
+                        .join(Shout)
+                        .join(Topic)
+                        .filter(Shout.slug == entry["slug"])
+                        .filter(Topic.slug == newslug)
+                        .first()
                     )
                     if not shout_topic_new:
                         try:
@@ -357,45 +336,37 @@ async def content_ratings_to_reactions(entry, slug):
             for content_rating in entry.get("ratings", []):
                 rater = (
                     session.query(User)
-                        .filter(User.oid == content_rating["createdBy"])
-                        .first()
+                    .filter(User.oid == content_rating["createdBy"])
+                    .first()
+                ) or User.default_user
+                shout = session.query(Shout).where(Shout.slug == slug).first()
+                cts = content_rating.get("createdAt")
+                reaction_dict = {
+                    "createdAt": date_parse(cts) if cts else None,
+                    "kind": ReactionKind.LIKE
+                    if content_rating["value"] > 0
+                    else ReactionKind.DISLIKE,
+                    "createdBy": rater.id,
+                    "shout": shout.id
+                }
+                reaction = (
+                    session.query(Reaction)
+                    .filter(Reaction.shout == reaction_dict["shout"])
+                    .filter(Reaction.createdBy == reaction_dict["createdBy"])
+                    .filter(Reaction.kind == reaction_dict["kind"])
+                    .first()
                 )
-                reactedBy = (
-                    rater
-                    if rater
-                    else session.query(User).filter(User.slug == "anonymous").first()
-                )
-                if rater:
-                    shout = session.query(Shout).where(Shout.slug == slug).one()
-
-                    reaction_dict = {
-                        "kind": ReactionKind.LIKE
-                        if content_rating["value"] > 0
-                        else ReactionKind.DISLIKE,
-                        "createdBy": reactedBy.id,
-                        "shout": shout.id,
-                    }
-                    cts = content_rating.get("createdAt")
-                    if cts:
-                        reaction_dict["createdAt"] = date_parse(cts)
-                    reaction = (
-                        session.query(Reaction).filter(
-                            Reaction.shout == reaction_dict["shout"]
-                        ).filter(
-                            Reaction.createdBy == reaction_dict["createdBy"]
-                        ).filter(
-                            Reaction.kind == reaction_dict["kind"]
-                        ).first()
-                    )
-                    if reaction:
-                        k = ReactionKind.AGREE if content_rating["value"] > 0 else ReactionKind.DISAGREE
-                        reaction_dict["kind"] = k
-                        reaction.update(reaction_dict)
-                    else:
-                        rea = Reaction.create(**reaction_dict)
-                        session.add(rea)
-                    # shout_dict['ratings'].append(reaction_dict)
+                if reaction:
+                    k = ReactionKind.AGREE if content_rating["value"] > 0 else ReactionKind.DISAGREE
+                    reaction_dict["kind"] = k
+                    reaction.update(reaction_dict)
+                    session.add(reaction)
+                else:
+                    rea = Reaction.create(**reaction_dict)
+                    session.add(rea)
+                    # await ReactedStorage.react(rea)
+                # shout_dict['ratings'].append(reaction_dict)
 
             session.commit()
     except Exception:
-        raise Exception("[migration] content_item.ratings error: \n%r" % content_rating)
+        print("[migration] content_item.ratings error: \n%r" % content_rating)