<commit_before>'''
Created on April 26, 2013

@package: livedesk
@copyright: 2013 Sourcefabric o.p.s.
@license: http://www.gnu.org/licenses/gpl-3.0.txt
@author: <NAME>

API implementation of liveblog sync.
'''

import socket
import json
import logging
import time
import codecs
from hashlib import sha1
from sched import scheduler
from threading import Thread
from urllib.request import urlopen, Request
from urllib.parse import urlparse, parse_qsl, urlencode, urlunparse
from datetime import datetime
from livedesk.api.blog_sync import IBlogSyncService, QBlogSync, BlogSync
from superdesk.source.api.source import ISourceService, Source, QSource
from livedesk.api.blog_post import BlogPost, IBlogPostService
from sqlalchemy.sql.functions import current_timestamp
from superdesk.collaborator.api.collaborator import ICollaboratorService, Collaborator
from ally.container import wire, app
from ally.container.ioc import injected
from ally.container.support import setup
from superdesk.user.api.user import IUserService, QUser, User
from ally.exception import InputError
from urllib.error import HTTPError
from time import sleep
from superdesk.media_archive.api.meta_data import IMetaDataUploadService
from superdesk.media_archive.api.meta_info import IMetaInfoService
from superdesk.person_icon.api.person_icon import IPersonIconService
from .icon_content import ChainedIconContent

# --------------------------------------------------------------------

log = logging.getLogger(__name__)

# --------------------------------------------------------------------

@injected
@setup(name='blogSynchronizer')
class BlogSyncProcess:
    '''
    Blog sync process.
    '''

    blogSyncService = IBlogSyncService; wire.entity('blogSyncService')
    # blog sync service used to retrieve blogs set on auto publishing

    sourceService = ISourceService; wire.entity('sourceService')
    # source service used to retrieve source data

    blogPostService = IBlogPostService; wire.entity('blogPostService')
    # blog post service used to insert blog posts

    collaboratorService = ICollaboratorService; wire.entity('collaboratorService')
    # blog post service used to retrive collaborator

    userService = IUserService; wire.entity('userService')

    metaDataService = IMetaDataUploadService; wire.entity('metaDataService')

    metaInfoService = IMetaInfoService; wire.entity('metaInfoService')

    personIconService = IPersonIconService; wire.entity('personIconService')

    syncThreads = {}
    # dictionary of threads that perform synchronization

    sync_interval = 10; wire.config('sync_interval', doc='''
    The number of seconds to perform sync for blogs.''')
    date_time_format = '%Y-%m-%d %H:%M:%S'; wire.config('date_time_format', doc='''
    The date time format used in REST requests.''')
    published_posts_path = 'Post/Published'; wire.config('published_posts_path', doc='''
    The partial path used to construct the URL for published posts retrieval''')

    acceptType = 'text/json'
    # mime type accepted for response from remote blog
    encodingType = 'UTF-8'
    # character encoding type accepted for response from remove blog

    @app.deploy
    def startSyncThread(self):
        '''
        Starts the sync thread.
        '''
        schedule = scheduler(time.time, time.sleep)
        def syncBlogs():
            self.syncBlogs()
            schedule.enter(self.sync_interval, 1, syncBlogs, ())
        schedule.enter(self.sync_interval, 1, syncBlogs, ())
        scheduleRunner = Thread(name='blogs sync', target=schedule.run)
        scheduleRunner.daemon = True
        scheduleRunner.start()
        log.info('Started the blogs automatic synchronization.')

    def syncBlogs(self):
        '''
        Read all blog sync entries for which auto was set true and sync
        the corresponding blogs.
        '''
        for blogSync in self.blogSyncService.getAll(q=QBlogSync(auto=True)):
            assert isinstance(blogSync, BlogSync)
            key = (blogSync.Blog, blogSync.Source)
            thread = self.syncThreads.get(key)
            if thread:
                assert isinstance(thread, Thread), 'Invalid thread %s' % thread
                if thread.is_alive(): continue

            self.syncThreads[key] = Thread(name='blog %d sync' % blogSync.Blog,
                                           target=self._syncBlog, args=(blogSync,))
            self.syncThreads[key].daemon = True
            self.syncThreads[key].start()
            log.info('Thread started for blog id %d and source id %d', blogSync.Blog, blogSync.Source)

    def _syncBlogLoop(self, blogSync):
        while True:
            log.info('Start sync for blog id %d and source id %d', blogSync.Blog, blogSync.Source)
            self._syncBlog(blogSync)
            sleep(self.sync_interval)

    def _syncBlog(self, blogSync):
        '''
        Synchronize the blog for the given sync entry.

        @param blogSync: BlogSync
            The blog sync entry declaring the blog and source from which the blog
            has to be updated.
        '''
        assert isinstance(blogSync, BlogSync), 'Invalid blog sync %s' % blogSync
        source = self.sourceService.getById(blogSync.Source)
        assert isinstance(source, Source)
        (scheme, netloc, path, params, query, fragment) = urlparse(source.URI)

        q = parse_qsl(query, keep_blank_values=True)
        q.append(('asc', 'cId'))
        q.append(('cId.since', blogSync.CId if blogSync.CId is not None else 0))
        if blogSync.SyncStart is not None:
            q.append(('publishedOn.since', blogSync.SyncStart.strftime(self.date_time_format)))
        url = urlunparse((scheme, netloc, path + '/' + self.published_posts_path, params, urlencode(q), fragment))
        req = Request(url, headers={'Accept' : self.acceptType, 'Accept-Charset' : self.encodingType,
                                    'X-Filter' : '*,Author.Source.*,Author.User.*', 'User-Agent' : 'Magic Browser'})
        try: resp = urlopen(req)
        except (HTTPError, socket.error) as e:
            log.error('Read error on %s: %s' % (source.URI, e))
            return
        if str(resp.status) != '200':
            log.error('Read problem on %s, status: %s' % (source.URI, resp.status))
            return

        try: msg = json.load(codecs.getreader(self.encodingType)(resp))
        except ValueError as e:
            log.error('Invalid JSON data %s' % e)
            return

        usersForIcons = {}
        for post in msg['PostList']:
            try:
                if post['IsPublished'] != 'True' or 'DeletedOn' in post: continue

                lPost = BlogPost()
                lPost.Type = post['Type']['Key']
                lPost.Creator = blogSync.Creator
                lPost.Author, userId = self._getCollaboratorForAuthor(post['Author'], source)
                lPost.Meta = post['Meta'] if 'Meta' in post else None
                lPost.ContentPlain = post['ContentPlain'] if 'ContentPlain' in post else None
                lPost.Content = post['Content'] if 'Content' in post else None
                lPost.CreatedOn = lPost.PublishedOn = current_timestamp()

                if userId and (userId not in usersForIcons):
                    try:
                        usersForIcons[userId] = post['Author']['User']
                    except KeyError:
                        pass

                # prepare the blog sync model to update the change identifier<commit_msg>LB-849 : Add the option to hide posts on collaborators, SMS, chained blogs and user comments<commit_after>'''
Created on April 26, 2013

@package: livedesk
@copyright: 2013 Sourcefabric o.p.s.
@license: http://www.gnu.org/licenses/gpl-3.0.txt
@author: <NAME>

API implementation of liveblog sync.
'''

import socket
import json
import logging
import time
import codecs
from hashlib import sha1
from sched import scheduler
from threading import Thread
from urllib.request import urlopen, Request
from urllib.parse import urlparse, parse_qsl, urlencode, urlunparse
from datetime import datetime
from livedesk.api.blog_sync import IBlogSyncService, QBlogSync, BlogSync
from superdesk.source.api.source import ISourceService, Source, QSource
from livedesk.api.blog_post import BlogPost, IBlogPostService
from sqlalchemy.sql.functions import current_timestamp
from superdesk.collaborator.api.collaborator import ICollaboratorService, Collaborator
from ally.container import wire, app
from ally.container.ioc import injected
from ally.container.support import setup
from superdesk.user.api.user import IUserService, QUser, User
from ally.exception import InputError
from urllib.error import HTTPError
from time import sleep
from superdesk.media_archive.api.meta_data import IMetaDataUploadService
from superdesk.media_archive.api.meta_info import IMetaInfoService
from superdesk.person_icon.api.person_icon import IPersonIconService
from .icon_content import ChainedIconContent

# --------------------------------------------------------------------

log = logging.getLogger(__name__)

# --------------------------------------------------------------------

@injected
@setup(name='blogSynchronizer')
class BlogSyncProcess:
    '''
    Blog sync process.
    '''

    blogSyncService = IBlogSyncService; wire.entity('blogSyncService')
    # blog sync service used to retrieve blogs set on auto publishing

    sourceService = ISourceService; wire.entity('sourceService')
    # source service used to retrieve source data

    blogPostService = IBlogPostService; wire.entity('blogPostService')
    # blog post service used to insert blog posts

    collaboratorService = ICollaboratorService; wire.entity('collaboratorService')
    # blog post service used to retrive collaborator

    userService = IUserService; wire.entity('userService')

    metaDataService = IMetaDataUploadService; wire.entity('metaDataService')

    metaInfoService = IMetaInfoService; wire.entity('metaInfoService')

    personIconService = IPersonIconService; wire.entity('personIconService')

    syncThreads = {}
    # dictionary of threads that perform synchronization

    sync_interval = 10; wire.config('sync_interval', doc='''
    The number of seconds to perform sync for blogs.''')
    date_time_format = '%Y-%m-%d %H:%M:%S'; wire.config('date_time_format', doc='''
    The date time format used in REST requests.''')
    published_posts_path = 'Post/Published'; wire.config('published_posts_path', doc='''
    The partial path used to construct the URL for published posts retrieval''')

    acceptType = 'text/json'
    # mime type accepted for response from remote blog
    encodingType = 'UTF-8'
    # character encoding type accepted for response from remove blog

    @app.deploy
    def startSyncThread(self):
        '''
        Starts the sync thread.
        '''
        schedule = scheduler(time.time, time.sleep)
        def syncBlogs():
            self.syncBlogs()
            schedule.enter(self.sync_interval, 1, syncBlogs, ())
        schedule.enter(self.sync_interval, 1, syncBlogs, ())
        scheduleRunner = Thread(name='blogs sync', target=schedule.run)
        scheduleRunner.daemon = True
        scheduleRunner.start()
        log.info('Started the blogs automatic synchronization.')

    def syncBlogs(self):
        '''
        Read all blog sync entries for which auto was set true and sync
        the corresponding blogs.
        '''
        for blogSync in self.blogSyncService.getAll(): #q=QBlogSync(auto=True)):
            assert isinstance(blogSync, BlogSync)
            key = (blogSync.Blog, blogSync.Source)
            thread = self.syncThreads.get(key)
            if thread:
                assert isinstance(thread, Thread), 'Invalid thread %s' % thread
                if thread.is_alive(): continue

            self.syncThreads[key] = Thread(name='blog %d sync' % blogSync.Blog,
                                           target=self._syncBlog, args=(blogSync,))
            self.syncThreads[key].daemon = True
            self.syncThreads[key].start()
            log.info('Thread started for blog id %d and source id %d', blogSync.Blog, blogSync.Source)

#     def _syncBlogLoop(self, blogSync):
#         while True:
#             log.info('Start sync for blog id %d and source id %d', blogSync.Blog, blogSync.Source)
#             self._syncBlog(blogSync)
#             sleep(self.sync_interval)

    def _syncBlog(self, blogSync):
        '''
        Synchronize the blog for the given sync entry.

        @param blogSync: BlogSync
            The blog sync entry declaring the blog and source from which the blog
            has to be updated.
        '''
        assert isinstance(blogSync, BlogSync), 'Invalid blog sync %s' % blogSync
        source = self.sourceService.getById(blogSync.Source)
        assert isinstance(source, Source)
        (scheme, netloc, path, params, query, fragment) = urlparse(source.URI)

        q = parse_qsl(query, keep_blank_values=True)
        q.append(('asc', 'cId'))
        q.append(('cId.since', blogSync.CId if blogSync.CId is not None else 0))
        if blogSync.SyncStart is not None:
            q.append(('publishedOn.since', blogSync.SyncStart.strftime(self.date_time_format)))
        url = urlunparse((scheme, netloc, path + '/' + self.published_posts_path, params, urlencode(q), fragment))
        req = Request(url, headers={'Accept' : self.acceptType, 'Accept-Charset' : self.encodingType,
                                    'X-Filter' : '*,Author.Source.*,Author.User.*', 'User-Agent' : 'Magic Browser'})
        try: resp = urlopen(req)
        except (HTTPError, socket.error) as e:
            log.error('Read error on %s: %s' % (source.URI, e))
            return
        if str(resp.status) != '200':
            log.error('Read problem on %s, status: %s' % (source.URI, resp.status))
            return

        try: msg = json.load(codecs.getreader(self.encodingType)(resp))
        except ValueError as e:
            log.error('Invalid JSON data %s' % e)
            return

        usersForIcons = {}
        for post in msg['PostList']:
            try:
                if post['IsPublished'] != 'True' or 'DeletedOn' in post: continue

                lPost = BlogPost()
                lPost.Type = post['Type']['Key']
                lPost.Creator = blogSync.Creator
                lPost.Author, userId = self._getCollaboratorForAuthor(post['Author'], source)
                lPost.Meta = post['Meta'] if 'Meta' in post else None
                lPost.ContentPlain = post['ContentPlain'] if 'ContentPlain' in post else None
                lPost.Content = post['Content'] if 'Content' in post else None
                lPost.CreatedOn = current_timestamp()
                
                if post['Auto'] == 'True':
                    lPost.PublishedOn = current_timestamp()
                else:
                    lPost.PublishedOn = None     

                if userId and (userId not in usersForIcons):
                    try:
                        usersForIcons[userId] = post['Author']['User']
                    except KeyError:
                        pass

                # prepare the blog sync model to update the change identifier
{'fraction_non_alphanumeric': 0.11275185403950909, 'fraction_numerical': 0.0030845967053882, 'mean_word_length': 2.5862555895504826, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6982673', 'n_tokens_mistral': 4206, 'n_tokens_neox': 4035, 'n_words': 1197}