"""
Scheduled indexaction jobs
"""
import logging
import html
import re
from sni.esi.scope import EsiScope
from sni.esi.token import esi_get_on_befalf_of, has_esi_scope
from sni.scheduler import scheduler
from sni.user.models import User
from .index import get_user_location
from .models import (
EsiMail,
EsiMailRecipient,
EsiSkillPoints,
EsiWalletBalance,
)
[docs]def format_mail_body(body: str) -> str:
"""
Formats a mail body by removing most of the HTML tags.
"""
body = body.replace("<br>", "\n")
body = re.sub(r"<.*?>", "", body)
body = html.unescape(body)
return body
[docs]def index_user_location(usr: User):
"""
Indexes a user's location, online status, and ship
"""
try:
location = get_user_location(usr, invalidate_token_on_4xx=True)
location.save()
except Exception as error:
logging.error(
"Could not index location of character %d (%s): %s",
usr.character_id,
usr.character_name,
str(error),
)
[docs]@scheduler.scheduled_job("interval", hours=1)
def index_users_location():
"""
Indexes all user's location
"""
for usr in User.objects():
if (
has_esi_scope(usr, EsiScope.ESI_LOCATION_READ_LOCATION_V1)
and has_esi_scope(usr, EsiScope.ESI_LOCATION_READ_ONLINE_V1)
and has_esi_scope(usr, EsiScope.ESI_LOCATION_READ_SHIP_TYPE_V1)
):
scheduler.add_job(index_user_location, args=(usr,))
[docs]def index_user_mails(usr: User):
"""
Pulls a character's email
"""
try:
character_id = usr.character_id
headers = esi_get_on_befalf_of(
f"latest/characters/{character_id}/mail",
character_id,
invalidate_token_on_4xx=True,
).data
except Exception as error:
logging.error(
"Could not list recent emails of character %d (%s): %s",
usr.character_id,
usr.character_name,
str(error),
)
return
for header in headers:
mail_id = header["mail_id"]
if EsiMail.objects(mail_id=mail_id).first() is not None:
break
try:
mail = esi_get_on_befalf_of(
f"latest/characters/{character_id}/mail/{mail_id}",
character_id,
invalidate_token_on_4xx=True,
).data
EsiMail(
body=format_mail_body(mail["body"]),
from_id=mail["from"],
mail_id=mail_id,
recipients=[
EsiMailRecipient(**raw) for raw in mail["recipients"]
],
subject=mail["subject"],
timestamp=mail["timestamp"],
).save()
except Exception as error:
logging.error(
"Could not index mail %d from character %d (%s): %s",
mail_id,
usr.character_id,
usr.character_name,
str(error),
)
[docs]@scheduler.scheduled_job("interval", hours=1)
def index_users_mails():
"""
Index all user emails.
"""
for usr in User.objects():
if has_esi_scope(usr, EsiScope.ESI_MAIL_READ_MAIL_V1):
scheduler.add_job(index_user_mails, args=(usr,))
[docs]def index_user_skillpoints(usr: User):
"""
Measures a user's skillpoints. See
:class:`sni.index.models.EsiSkillPoints`.
"""
try:
data = esi_get_on_befalf_of(
f"latest/characters/{usr.character_id}/skills/",
usr.character_id,
invalidate_token_on_4xx=True,
).data
EsiSkillPoints(
total_sp=data["total_sp"],
unallocated_sp=data.get("unallocated_sp", 0),
user=usr,
).save()
except Exception as error:
logging.error(
"Could not index skillpoints of character %d (%s): %s",
usr.character_id,
usr.character_name,
str(error),
)
[docs]@scheduler.scheduled_job("interval", hours=12)
def index_users_skillpoints():
"""
Measures all users skillpoints
"""
for usr in User.objects():
if has_esi_scope(usr, EsiScope.ESI_SKILLS_READ_SKILLS_V1):
scheduler.add_job(index_user_skillpoints, args=(usr,))
[docs]def index_user_wallets(usr: User):
"""
Indexes user wallet balance
"""
try:
balance = esi_get_on_befalf_of(
f"latest/characters/{usr.character_id}/wallet/",
usr.character_id,
invalidate_token_on_4xx=True,
).data
EsiWalletBalance(balance=balance, user=usr).save()
except Exception as error:
logging.error(
"Could not index wallet of character %d (%s): %s",
usr.character_id,
usr.character_name,
str(error),
)
[docs]@scheduler.scheduled_job("interval", hours=12)
def index_users_wallets():
"""
Indexes user wallet balance
"""
for usr in User.objects():
if has_esi_scope(usr, EsiScope.ESI_WALLET_READ_CHARACTER_WALLET_V1):
scheduler.add_job(index_user_wallets, args=(usr,))