<commit_before>from tapiriik.database import db, cachedb, redis
from tapiriik.messagequeue import mq
from tapiriik.services import Service, ServiceRecord, APIExcludeActivity, ServiceException, ServiceExceptionScope, ServiceWarning, UserException, UserExceptionType
from tapiriik.settings import USER_SYNC_LOGS, DISABLED_SERVICES, WITHDRAWN_SERVICES
from .activity_record import ActivityRecord, ActivityServicePrescence
from datetime import datetime, timedelta
from pymongo.read_preferences import ReadPreference
import sys
import os
import socket
import traceback
import pprint
import copy
import random
import logging
import logging.handlers
import pytz
import kombu
import json

# Set this up seperate from the logger used in this scope, so services logging messages are caught and logged into user's files.
_global_logger = logging.getLogger("tapiriik")

_global_logger.setLevel(logging.DEBUG)
logging_console_handler = logging.StreamHandler(sys.stdout)
logging_console_handler.setLevel(logging.DEBUG)
logging_console_handler.setFormatter(logging.Formatter('%(message)s'))
_global_logger.addHandler(logging_console_handler)

logger = logging.getLogger("tapiriik.sync.worker")

def _formatExc():
    try:
        exc_type, exc_value, exc_traceback = sys.exc_info()
        tb = exc_traceback
        while tb.tb_next:
            tb = tb.tb_next
        frame = tb.tb_frame
        locals_trimmed = []
        for local_name, local_val in frame.f_locals.items():
            value_full = pprint.pformat(local_val)
            if len(value_full) > 1000:
                value_full = value_full[:500] + "..." + value_full[-500:]
            locals_trimmed.append(str(local_name) + "=" + value_full)
        exc = '\n'.join(traceback.format_exception(exc_type, exc_value, exc_traceback)) + "\nLOCALS:\n" + '\n'.join(locals_trimmed)
        logger.exception("Service exception")
        return exc
    finally:
        del exc_traceback, exc_value, exc_type

def _isWarning(exc):
    return issubclass(exc.__class__, ServiceWarning)

# It's practically an ORM!

def _packServiceException(step, e):
    res = {"Step": step, "Message": e.Message + "\n" + _formatExc(), "Block": e.Block, "Scope": e.Scope, "TriggerExhaustive": e.TriggerExhaustive, "Timestamp": datetime.utcnow()}
    if e.UserException:
        res["UserException"] = _packUserException(e.UserException)
    return res

def _packUserException(userException):
    if userException:
        return {"Type": userException.Type, "Extra": userException.Extra, "InterventionRequired": userException.InterventionRequired, "ClearGroup": userException.ClearGroup}

def _unpackUserException(raw):
    if not raw:
        return None
    if "UserException" in raw:
        raw = raw["UserException"]
    if not raw:
        return None
    if "Type" not in raw:
        return None
    return UserException(raw["Type"], extra=raw["Extra"], intervention_required=raw["InterventionRequired"], clear_group=raw["ClearGroup"])

class Sync:

    SyncInterval = timedelta(hours=1)
    SyncIntervalJitter = timedelta(minutes=5)
    MinimumSyncInterval = timedelta(seconds=30)
    MaximumIntervalBeforeExhaustiveSync = timedelta(days=14)  # Based on the general page size of 50 activites, this would be >3/day...

    def ScheduleImmediateSync(user, exhaustive=None):
        if exhaustive is None:
            db.users.update({"_id": user["_id"]}, {"$set": {"NextSynchronization": datetime.utcnow()}})
        else:
            db.users.update({"_id": user["_id"]}, {"$set": {"NextSynchronization": datetime.utcnow(), "NextSyncIsExhaustive": exhaustive}})

    def SetNextSyncIsExhaustive(user, exhaustive=False):
        db.users.update({"_id": user["_id"]}, {"$set": {"NextSyncIsExhaustive": exhaustive}})

    def InitializeWorkerBindings():
        Sync._channel = mq.channel()
        Sync._exchange = kombu.Exchange("tapiriik-users", type="direct")(Sync._channel)
        Sync._exchange.declare()
        Sync._global_queue = kombu.Queue("tapiriik-users")(Sync._channel)
        Sync._host_queue = kombu.Queue("tapiriik-users-%s" % socket.gethostname())(Sync._channel)
        Sync._global_queue.declare()
        Sync._host_queue.declare()
        # Bind to worker-specific and general routing keys
        Sync._global_queue.bind_to(exchange="tapiriik-users", routing_key="")
        Sync._host_queue.bind_to(exchange="tapiriik-users", routing_key=socket.gethostname())

    def PerformGlobalSync(heartbeat_callback=None, version=None, max_users=None):
        def _callback(body, message):
            Sync._consumeSyncTask(body, message, heartbeat_callback, version)

        Sync._consumer = kombu.Consumer(
            channel=Sync._channel,
            queues=[Sync._host_queue, Sync._global_queue],
            callbacks=[_callback],
            auto_declare=False
        )

        Sync._consumer.qos(prefetch_count=1, apply_global=False)

        Sync._consumer.consume()

        for _ in kombu.eventloop(mq, limit=max_users):
            pass

    def _consumeSyncTask(body, message, heartbeat_callback_direct, version):
        from tapiriik.auth import User

        user_id = body["user_id"]
        user = User.Get(user_id)
        if user is None:
            logger.warning("Could not find user %s - bailing" % user_id)
            message.ack() # Otherwise the entire thing grinds to a halt
            return
        if body["generation"] != user["QueuedGeneration"]:
            # QueuedGeneration being different means they've gone through sync_scheduler since this particular message was queued
            # So, discard this and wait for that message to surface
            # Should only happen when I manually requeue people
            logger.warning("Queue generation mismatch for %s - bailing" % user_id)
            message.ack()
            return

        def heartbeat_callback(state):
            heartbeat_callback_direct(state, user_id)

        syncStart = datetime.utcnow()

        # Always to an exhaustive sync if there were errors
        #   Sometimes services report that uploads failed even when they succeeded.
        #   If a partial sync was done, we'd be assuming that the accounts were consistent past the first page
        #       e.g. If an activity failed to upload far in the past, it would never be attempted again.
        #   So we need to verify the full state of the accounts.
        # But, we can still do a partial sync if there are *only* blocking errors
        #   In these cases, the block will protect that service from being improperly manipulated (though tbqh I can't come up with a situation where this would happen, it's more of a performance thing).
        #   And, when the block is cleared, NextSyncIsExhaustive is set.

        exhaustive = "NextSyncIsExhaustive" in user and user["NextSyncIsExhaustive"] is True
        if  ("ForcingExhaustiveSyncErrorCount" not in user and "NonblockingSyncErrorCount" in user and user["NonblockingSyncErrorCount"] > 0) or \<commit_msg>Stop users without QueuedGeneration crashing sync worker<commit_after>from tapiriik.database import db, cachedb, redis
from tapiriik.messagequeue import mq
from tapiriik.services import Service, ServiceRecord, APIExcludeActivity, ServiceException, ServiceExceptionScope, ServiceWarning, UserException, UserExceptionType
from tapiriik.settings import USER_SYNC_LOGS, DISABLED_SERVICES, WITHDRAWN_SERVICES
from .activity_record import ActivityRecord, ActivityServicePrescence
from datetime import datetime, timedelta
from pymongo.read_preferences import ReadPreference
import sys
import os
import socket
import traceback
import pprint
import copy
import random
import logging
import logging.handlers
import pytz
import kombu
import json

# Set this up seperate from the logger used in this scope, so services logging messages are caught and logged into user's files.
_global_logger = logging.getLogger("tapiriik")

_global_logger.setLevel(logging.DEBUG)
logging_console_handler = logging.StreamHandler(sys.stdout)
logging_console_handler.setLevel(logging.DEBUG)
logging_console_handler.setFormatter(logging.Formatter('%(message)s'))
_global_logger.addHandler(logging_console_handler)

logger = logging.getLogger("tapiriik.sync.worker")

def _formatExc():
    try:
        exc_type, exc_value, exc_traceback = sys.exc_info()
        tb = exc_traceback
        while tb.tb_next:
            tb = tb.tb_next
        frame = tb.tb_frame
        locals_trimmed = []
        for local_name, local_val in frame.f_locals.items():
            value_full = pprint.pformat(local_val)
            if len(value_full) > 1000:
                value_full = value_full[:500] + "..." + value_full[-500:]
            locals_trimmed.append(str(local_name) + "=" + value_full)
        exc = '\n'.join(traceback.format_exception(exc_type, exc_value, exc_traceback)) + "\nLOCALS:\n" + '\n'.join(locals_trimmed)
        logger.exception("Service exception")
        return exc
    finally:
        del exc_traceback, exc_value, exc_type

def _isWarning(exc):
    return issubclass(exc.__class__, ServiceWarning)

# It's practically an ORM!

def _packServiceException(step, e):
    res = {"Step": step, "Message": e.Message + "\n" + _formatExc(), "Block": e.Block, "Scope": e.Scope, "TriggerExhaustive": e.TriggerExhaustive, "Timestamp": datetime.utcnow()}
    if e.UserException:
        res["UserException"] = _packUserException(e.UserException)
    return res

def _packUserException(userException):
    if userException:
        return {"Type": userException.Type, "Extra": userException.Extra, "InterventionRequired": userException.InterventionRequired, "ClearGroup": userException.ClearGroup}

def _unpackUserException(raw):
    if not raw:
        return None
    if "UserException" in raw:
        raw = raw["UserException"]
    if not raw:
        return None
    if "Type" not in raw:
        return None
    return UserException(raw["Type"], extra=raw["Extra"], intervention_required=raw["InterventionRequired"], clear_group=raw["ClearGroup"])

class Sync:

    SyncInterval = timedelta(hours=1)
    SyncIntervalJitter = timedelta(minutes=5)
    MinimumSyncInterval = timedelta(seconds=30)
    MaximumIntervalBeforeExhaustiveSync = timedelta(days=14)  # Based on the general page size of 50 activites, this would be >3/day...

    def ScheduleImmediateSync(user, exhaustive=None):
        if exhaustive is None:
            db.users.update({"_id": user["_id"]}, {"$set": {"NextSynchronization": datetime.utcnow()}})
        else:
            db.users.update({"_id": user["_id"]}, {"$set": {"NextSynchronization": datetime.utcnow(), "NextSyncIsExhaustive": exhaustive}})

    def SetNextSyncIsExhaustive(user, exhaustive=False):
        db.users.update({"_id": user["_id"]}, {"$set": {"NextSyncIsExhaustive": exhaustive}})

    def InitializeWorkerBindings():
        Sync._channel = mq.channel()
        Sync._exchange = kombu.Exchange("tapiriik-users", type="direct")(Sync._channel)
        Sync._exchange.declare()
        Sync._global_queue = kombu.Queue("tapiriik-users")(Sync._channel)
        Sync._host_queue = kombu.Queue("tapiriik-users-%s" % socket.gethostname())(Sync._channel)
        Sync._global_queue.declare()
        Sync._host_queue.declare()
        # Bind to worker-specific and general routing keys
        Sync._global_queue.bind_to(exchange="tapiriik-users", routing_key="")
        Sync._host_queue.bind_to(exchange="tapiriik-users", routing_key=socket.gethostname())

    def PerformGlobalSync(heartbeat_callback=None, version=None, max_users=None):
        def _callback(body, message):
            Sync._consumeSyncTask(body, message, heartbeat_callback, version)

        Sync._consumer = kombu.Consumer(
            channel=Sync._channel,
            queues=[Sync._host_queue, Sync._global_queue],
            callbacks=[_callback],
            auto_declare=False
        )

        Sync._consumer.qos(prefetch_count=1, apply_global=False)

        Sync._consumer.consume()

        for _ in kombu.eventloop(mq, limit=max_users):
            pass

    def _consumeSyncTask(body, message, heartbeat_callback_direct, version):
        from tapiriik.auth import User

        user_id = body["user_id"]
        user = User.Get(user_id)
        if user is None:
            logger.warning("Could not find user %s - bailing" % user_id)
            message.ack() # Otherwise the entire thing grinds to a halt
            return
        if body["generation"] != user.get("QueuedGeneration", None):
            # QueuedGeneration being different means they've gone through sync_scheduler since this particular message was queued
            # So, discard this and wait for that message to surface
            # Should only happen when I manually requeue people
            logger.warning("Queue generation mismatch for %s - bailing" % user_id)
            message.ack()
            return

        def heartbeat_callback(state):
            heartbeat_callback_direct(state, user_id)

        syncStart = datetime.utcnow()

        # Always to an exhaustive sync if there were errors
        #   Sometimes services report that uploads failed even when they succeeded.
        #   If a partial sync was done, we'd be assuming that the accounts were consistent past the first page
        #       e.g. If an activity failed to upload far in the past, it would never be attempted again.
        #   So we need to verify the full state of the accounts.
        # But, we can still do a partial sync if there are *only* blocking errors
        #   In these cases, the block will protect that service from being improperly manipulated (though tbqh I can't come up with a situation where this would happen, it's more of a performance thing).
        #   And, when the block is cleared, NextSyncIsExhaustive is set.

        exhaustive = "NextSyncIsExhaustive" in user and user["NextSyncIsExhaustive"] is True
        if  ("ForcingExhaustiveSyncErrorCount" not in user and "NonblockingSyncErrorCount" in user and user["NonblockingSyncErrorCount"] > 0) or \
{'fraction_non_alphanumeric': 0.09538045412484875, 'fraction_numerical': 0.0029895366218236174, 'mean_word_length': 3.496, 'pattern_counts': {'":': 40, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3475958', 'n_tokens_mistral': 4099, 'n_tokens_neox': 3902, 'n_words': 1101}