src/depot/publishers/woblink.py

   1 from datetime import date
   2 import io
   3 import json
   4 import re
   5 from time import sleep
   6 from django.conf import settings
   7 from django.utils.html import escape, format_html
   8 from django.utils.safestring import mark_safe
   9 from librarian.builders.html import SnippetHtmlBuilder
  10 from librarian.functions import lang_code_3to2
  11 from catalogue.models import Audience, Author, Thema
  12 from .. import models
  13 from .base import BasePublisher
  14 from .woblink_constants import WOBLINK_CATEGORIES
  15
  16
  17 class WoblinkError(ValueError):
  18     pass
  19
  20 class NoPrice(WoblinkError):
  21     def as_html(self):
  22         return format_html(
  23             'Brak <a href="/admin/depot/shop/{price}">określonej ceny</a>.',
  24             price=self.args[0].id
  25         )
  26
  27 class NoIsbn(WoblinkError):
  28     def as_html(self):
  29         return 'Brak ISBN.'
  30
  31 class AuthorLiteralForeign(WoblinkError):
  32     def as_html(self):
  33         return format_html(
  34             'Nie obsługiwane: autor „{author}” w języku {lang}.',
  35             author=str(self.args[0]),
  36             lang=self.args[0].lang,
  37         )
  38
  39 class AuthorNotInCatalogue(WoblinkError):
  40     def as_html(self):
  41         return format_html(
  42             'Brak autora „{author}” w katalogu.',
  43             author=str(self.args[0])
  44         )
  45
  46 class AuthorNoWoblink(WoblinkError):
  47     def as_html(self):
  48         return format_html(
  49             'Autor <a href="/admin/catalogue/author/{author_id}/">{author}</a> bez identyfikatora Woblink.',
  50             author_id=self.args[0].id,
  51             author=self.args[0].name
  52         )
  53
  54 class NoThema(WoblinkError):
  55     def as_html(self):
  56         return format_html('Brak Thema.')
  57
  58 class UnknownThema(WoblinkError):
  59     def as_html(self):
  60         return format_html(
  61             'Nieznana Thema {code}.',
  62             code=self.args[0]
  63         )
  64
  65
  66 class ThemaUnknownWoblink(WoblinkError):
  67     def as_html(self):
  68         return format_html(
  69             'Thema <a href="/admin/catalogue/thema/{id}/">{code}</a> przypisana do nieznanej kategorii Woblink.',
  70             id=self.args[0].id,
  71             code=self.args[0].code,
  72         )
  73
  74 class NoWoblinkCategory(WoblinkError):
  75     def as_html(self):
  76         return 'Brak kategorii Woblink.'
  77
  78 class WoblinkWarning(Warning):
  79     pass
  80
  81 class NoMainThemaWarning(WoblinkWarning):
  82     def as_html(self):
  83         return format_html(
  84             'Brak głównej kategorii Thema.'
  85         )
  86
  87 class ThemaNoWoblink(WoblinkWarning):
  88     def as_html(self):
  89         return format_html(
  90             'Thema <a href="/admin/catalogue/thema/{id}/">{code}</a> nie przypisana do kategorii Woblink.',
  91             id=self.args[0].id,
  92             code=self.args[0].code,
  93         )
  94
  95 class AuthorLiteralForeignWarning(WoblinkWarning):
  96     def as_html(self):
  97         return format_html(
  98             'Nie obsługiwane: autor „{author}” w języku {lang}.',
  99             author=str(self.args[0]),
 100             lang=self.args[0].lang,
 101         )
 102
 103 class AuthorNotInCatalogueWarning(WoblinkWarning):
 104     def as_html(self):
 105         return format_html(
 106             'Brak autora „{author}” w katalogu.',
 107             author=str(self.args[0])
 108         )
 109
 110 class AuthorNoWoblinkWarning(WoblinkWarning):
 111     def as_html(self):
 112         return format_html(
 113             'Autor <a href="/admin/catalogue/author/{author_id}/">{author}</a> bez identyfikatora Woblink.',
 114             author_id=self.args[0].id,
 115             author=self.args[0].name
 116         )
 117
 118
 119
 120
 121 class Woblink(BasePublisher):
 122     BASE_URL = 'https://publisher.woblink.com/'
 123     ADD_URL = BASE_URL + 'catalog/add'
 124     STEP1_URL = BASE_URL + 'catalog/edit/%s'
 125     STEP2_URL = BASE_URL + 'catalog/edit/%s/2'
 126     STEP3_URL = BASE_URL + 'catalog/edit/%s/3'
 127     UPLOAD_URL = BASE_URL + 'file/upload-%s'
 128     JOB_STATUS_URL = BASE_URL + 'task/status'
 129     GENERATE_DEMO_URL = BASE_URL + 'task/run/generate-%s-demo/%s/%d'
 130     CHECK_DEMO_URL = BASE_URL + 'task/run/check-%s-demo/%s'
 131
 132     SEARCH_CATALOGUE_URL = BASE_URL + '{category}/autocomplete/{term}'
 133
 134     ROLE_AUTHOR = 1
 135     ROLE_TRANSLATOR = 4
 136
 137     def login(self):
 138         response = self.session.get('https://publisher.woblink.com/login')
 139         token = re.search(
 140             r'name="_csrf_token" value="([^"]+)"',
 141             response.text
 142         ).group(1)
 143         data = {
 144             '_csrf_token': token,
 145             '_username': self.username,
 146             '_password': self.password,
 147         }
 148         response = self.session.post(
 149             'https://publisher.woblink.com/login_check',
 150             data=data,
 151         )
 152
 153     def search_catalogue(self, category, term):
 154         return self.session.get(
 155             self.SEARCH_CATALOGUE_URL.format(category=category, term=term)
 156         ).json()
 157
 158     def search_author_catalogue(self, term):
 159         return [
 160             {
 161                 'id': item['autId'],
 162                 'text': item['autFullname']
 163             }
 164             for item in self.search_catalogue('author', term)
 165         ]
 166     def search_series_catalogue(self, term):
 167         return [
 168             {
 169                 'id': item['id'],
 170                 'text': item['name']
 171             }
 172             for item in self.search_catalogue('series', term)
 173         ]
 174
 175     def get_isbn(self, meta, errors=None):
 176         if not meta.isbn_epub:
 177             if errors is not None:
 178                 errors.append(NoIsbn())
 179         return meta.isbn_epub
 180
 181     def get_authors_data(self, meta, errors=None):
 182         authors = []
 183         for role, items, obligatory in [
 184                 (self.ROLE_AUTHOR, meta.authors, True),
 185                 (self.ROLE_TRANSLATOR, meta.translators, False)
 186         ]:
 187             for person_literal in items:
 188                 if person_literal.lang != 'pl':
 189                     if errors is not None:
 190                         if obligatory:
 191                              errors.append(AuthorLiteralForeign(person_literal))
 192                         else:
 193                             errors.append(AuthorLiteralForeignWarning(person_literal))
 194                     continue
 195                 aobj = Author.get_by_literal(str(person_literal))
 196                 if aobj is None:
 197                     if errors is not None:
 198                         if obligatory:
 199                              errors.append(AuthorNotInCatalogue(person_literal))
 200                         else:
 201                             errors.append(AuthorNotInCatalogueWarning(person_literal))
 202                     continue
 203                 if not aobj.woblink:
 204                     if errors is not None:
 205                         if obligatory:
 206                              errors.append(AuthorNoWoblink(aobj))
 207                         else:
 208                             errors.append(AuthorNoWoblinkWarning(aobj))
 209                     continue
 210                 authors.append((role, aobj.woblink))
 211         return authors
 212
 213     def get_genres(self, meta, errors=None):
 214         thema_codes = []
 215         if meta.thema_main:
 216             thema_codes.append(meta.thema_main)
 217         else:
 218             if errors is not None:
 219                 errors.append(NoMainThemaWarning())
 220         thema_codes.extend(meta.thema)
 221         if not thema_codes:
 222             if errors is not None:
 223                 errors.append(NoThema())
 224         category_ids = []
 225         for code in thema_codes:
 226             try:
 227                 thema = Thema.objects.get(code=code)
 228             except Thema.DoesNotExist:
 229                 if errors is not None:
 230                     errors.append(UnknownThema(code))
 231             else:
 232                 if thema.woblink_category is None:
 233                     if errors is not None:
 234                         errors.append(ThemaNoWoblink(thema))
 235                 elif thema.woblink_category not in WOBLINK_CATEGORIES:
 236                     if errors is not None:
 237                         errors.append(ThemaUnknownWoblink(thema))
 238                 elif thema.woblink_category not in category_ids:
 239                     category_ids.append(thema.woblink_category)
 240         if not category_ids:
 241             if errors is not None:
 242                 errors.append(NoWoblinkCategory())
 243         return category_ids
 244
 245     def get_series(self, meta, errors=None):
 246         return list(Audience.objects.filter(code__in=audiences).exclude(
 247             woblink=None).values_list('woblink', flat=True))
 248
 249     def get_abstract(self, wldoc, errors=None, description_add=None):
 250         description = self.get_description(wldoc, description_add)
 251         parts = description.split('\n', 1)
 252         if len(parts) == 1 or len(parts[0]) > 200:
 253             p1 = description[:200].rsplit(' ', 1)[0]
 254             p2 = description[len(p1):]
 255             p1 += '…'
 256             p2 = '…' + p2
 257             parts = [p1, p2]
 258
 259         m = re.search(r'<[^>]+$', parts[0])
 260         if m is not None:
 261             parts[0] = parts[:-len(m.group(0))]
 262             parts[1] = m.group(0) + parts[1]
 263
 264         opened = []
 265         for tag in re.findall(r'<[^>]+[^/>]>', parts[0]):
 266             if tag[1] == '/':
 267                 opened.pop()
 268             else:
 269                 opened.append(tag)
 270         for tag in reversed(opened):
 271             parts[0] += '</' + tag[1:-1].split()[0] + '>'
 272             parts[1] = tag + parts[1]
 273         return {
 274             'header': parts[0],
 275             'rest': parts[1],
 276         }
 277
 278     def get_lang2code(self, meta, errors=None):
 279         return lang_code_3to2(meta.language)
 280
 281     def get_price(self, shop, wldoc, errors=None):
 282         stats = wldoc.get_statistics()['total']
 283         words = stats['words_with_fn']
 284         pages = stats['chars_with_fn'] / 1800
 285         price = shop.get_price(words, pages)
 286         if price is None:
 287             if errors:
 288                 errors.append(NoPrice(shop))
 289             return 0
 290
 291         return price
 292
 293     def can_publish(self, shop, book):
 294         wldoc = book.wldocument(librarian2=True)
 295         d = {
 296             'warnings': [],
 297             'errors': [],
 298         }
 299         errors = []
 300         book_data = self.get_book_data(shop, wldoc, errors)
 301         for error in errors:
 302             if not isinstance(error, Warning):
 303                 errlist = d['errors']
 304             else:
 305                 errlist = d['warnings']
 306             errlist.append(error.as_html())
 307
 308         if book_data.get('genres'):
 309             d['comment'] = format_html(
 310                 'W kategoriach: {cat} ({price} zł)',
 311                 cat=', '.join(self.describe_category(g) for g in book_data['genres']),
 312                 price=book_data['price']
 313             )
 314
 315         return d
 316
 317     def describe_category(self, category):
 318         t = []
 319         while category:
 320             c = WOBLINK_CATEGORIES[category]
 321             t.append(c['name'])
 322             category = c.get('parent')
 323         return ' / '.join(reversed(t))
 324
 325     def create_book(self, isbn):
 326         isbn = ''.join(c for c in isbn if c.isdigit())
 327         assert len(isbn) == 13
 328         response = self.session.post(
 329             self.ADD_URL,
 330             data={
 331                 'AddPublication[pubType]': 'ebook',
 332                 'AddPublication[pubHasIsbn]': '1',
 333                 'AddPublication[pubIsbn]': isbn,
 334                  ##AddPubation[save]
 335             }
 336         )
 337         m = re.search(r'/(\d+)$', response.url)
 338         if m is not None:
 339             return m.group(1)
 340
 341     def send_book(self, shop, book, changes=None):
 342         wldoc = book.wldocument(librarian2=True, changes=changes, publishable=False) # TODO pub
 343         meta = wldoc.meta
 344
 345         book_data = self.get_book_data(shop, wldoc)
 346
 347         if not book.woblink_id:
 348             #book.woblink_id = 2959868
 349             woblink_id = self.create_book(book_data['isbn'])
 350             assert woblink_id
 351             book.woblink_id = woblink_id
 352             book.save(update_fields=['woblink_id'])
 353
 354         self.edit_step1(book.woblink_id, book_data)
 355         self.edit_step2(book.woblink_id, book_data)
 356         self.edit_step3(book.woblink_id, book_data)
 357         self.send_cover(book.woblink_id, wldoc)
 358         texts = shop.get_texts()
 359         self.send_epub(
 360             book.woblink_id, wldoc, book.gallery_path(),
 361             fundraising=texts
 362         )
 363         self.send_mobi(
 364             book.woblink_id, wldoc, book.gallery_path(),
 365             fundraising=texts
 366         )
 367
 368     def get_book_data(self, shop, wldoc, errors=None):
 369         return {
 370             "title": wldoc.meta.title,
 371             "isbn": self.get_isbn(wldoc.meta, errors=errors),
 372             "authors": self.get_authors_data(wldoc.meta, errors=errors),
 373             "abstract": self.get_abstract(
 374                 wldoc, errors=errors, description_add=shop.description_add
 375             ),
 376             "lang2code": self.get_lang2code(wldoc.meta, errors=errors),
 377             "genres": self.get_genres(wldoc.meta, errors=errors),
 378             "price": self.get_price(shop, wldoc, errors=errors),
 379             "series": self.get_series(wldoc.meta, errors=errors),
 380         }
 381
 382     def with_form_name(self, data, name):
 383         return {
 384             f"{name}[{k}]": v
 385             for (k, v) in data.items()
 386         }
 387
 388     def edit_step1(self, woblink_id, book_data):
 389         data = book_data
 390
 391         authors_data = [
 392             {
 393                 "AhpPubId": woblink_id,
 394                 "AhpAutId": author_id,
 395                 "AhpType": author_type,
 396             }
 397             for (author_type, author_id) in data['authors']
 398         ]
 399
 400         series_data = [
 401             {
 402                 'PublicationId': woblink_id,
 403                 'SeriesId': series_id,
 404             }
 405             for series_id in data['series']
 406         ]
 407
 408         d = {
 409             'pubTitle': book_data['title'],
 410             'npwAuthorHasPublications': json.dumps(authors_data),
 411             'pubShortNote': data['abstract']['header'],
 412             'pubNote': data['abstract']['rest'],
 413             'pubCulture': data['lang2code'],
 414             'npwPublicationHasAwards': '[]',
 415             'npwPublicationHasSeriess': json.dumps(series_id),
 416                 # "[{\"Id\":6153,\"PublicationId\":73876,\"SeriesId\":1615,\"Tome\":null}]"
 417         }
 418         d = self.with_form_name(d, 'EditPublicationStep1')
 419         d['roles'] = [author_type for (author_type, author_id) in data['authors']]
 420         r = self.session.post(self.STEP1_URL % woblink_id, data=d)
 421         return r
 422
 423
 424     def edit_step2(self, woblink_id, book_data):
 425         gd = {}
 426         legacy = None
 427         for i, g in enumerate(book_data['genres']):
 428             gdata = WOBLINK_CATEGORIES[g]
 429             if legacy is None:
 430                 legacy = gdata.get('legacy')
 431             if p := gdata.get('parent'):
 432                 gd.setdefault(p, {'isMain': False})
 433                 gd[p].setdefault('children', [])
 434                 gd[p]['children'].append(str(g))
 435                 gd[p].setdefault('mainChild', str(g))
 436                 if legacy is None:
 437                     legacy = WOBLINK_CATEGORIES[p].get('legacy')
 438             else:
 439                 gd.setdefault(g, {})
 440                 gd[g]['isMain'] = True
 441         gd = [
 442             {
 443                 "pubId": woblink_id,
 444                 "category": str(k),
 445                 **v
 446             }
 447             for k, v in gd.items()
 448         ]
 449
 450         data = {
 451             'npwPublicationHasNewGenres': json.dumps(gd),
 452             'genre': legacy or '',
 453         }
 454         data = self.with_form_name(data, 'AddPublicationStep2')
 455         return self.session.post(self.STEP2_URL % woblink_id, data=data)
 456
 457     def edit_step3(self, woblink_id, book_data):
 458         d = {
 459             'pubBasePrice': book_data['price'],
 460             'pubPremiereDate': date.today().isoformat(),
 461             'pubIsLicenseIndefinite': '1',
 462             'pubFileFormat': 'epub+mobi',
 463             'pubIsAcs': '0',
 464             'pubPublisherIndex': '',
 465         }
 466         d = self.with_form_name(d, 'EditPublicationStep3')
 467         return self.session.post(self.STEP3_URL % woblink_id, data=d)
 468
 469     def wait_for_job(self, job_id):
 470         while True:
 471             response = self.session.post(
 472                 self.JOB_STATUS_URL,
 473                 data={'ids[]': job_id}
 474             )
 475             data = response.json()[job_id]
 476             if data['ready']:
 477                 assert data['successful']
 478                 return
 479             sleep(2)
 480
 481     def upload_file(self, woblink_id, filename, content, form_name, field_name, mime_type):
 482         data = {
 483             'pubId': woblink_id,
 484         }
 485         files = {
 486             field_name: (filename, content, mime_type)
 487         }
 488         response = self.session.post(
 489             self.UPLOAD_URL % field_name,
 490             data=self.with_form_name(data, form_name),
 491             files=self.with_form_name(files, form_name),
 492         )
 493         resp_data = response.json()
 494         assert resp_data['success'] is True
 495         if 'jobId' in resp_data:
 496             self.wait_for_job(resp_data['jobId'])
 497
 498     def generate_demo(self, woblink_id, file_format, check=True):
 499         percent = 10
 500         while True:
 501             job_id = self.session.get(
 502                 self.GENERATE_DEMO_URL % (file_format, woblink_id, percent),
 503             ).json()['jobId']
 504             try:
 505                 self.wait_for_job(job_id)
 506             except AssertionError:
 507                 if percent < 50:
 508                     percent += 10
 509                 else:
 510                     raise
 511             else:
 512                 break
 513
 514         if check:
 515             self.wait_for_job(
 516                 self.session.get(
 517                     self.CHECK_DEMO_URL % (file_format, woblink_id)
 518                 ).json()['jobId']
 519             )
 520
 521     def send_epub(self, woblink_id, doc, gallery_path, fundraising=None):
 522         from librarian.builders import EpubBuilder
 523         content = EpubBuilder(
 524             base_url='file://' + gallery_path + '/',
 525             fundraising=fundraising or [],
 526         ).build(doc).get_file()
 527         self.upload_file(
 528             woblink_id,
 529             doc.meta.url.slug + '.epub',
 530             content,
 531             'UploadEpub',
 532             'epub',
 533             'application/epub+zip'
 534         )
 535         self.generate_demo(woblink_id, 'epub')
 536
 537     def send_mobi(self, woblink_id, doc, gallery_path, fundraising=None):
 538         from librarian.builders import MobiBuilder
 539         content = MobiBuilder(
 540             base_url='file://' + gallery_path + '/',
 541             fundraising=fundraising or [],
 542         ).build(doc).get_file()
 543         self.upload_file(
 544             woblink_id,
 545             doc.meta.url.slug + '.mobi',
 546             content,
 547             'UploadMobi',
 548             'mobi',
 549             'application/x-mobipocket-ebook'
 550         )
 551         self.generate_demo(woblink_id, 'mobi', check=False)
 552
 553     def send_cover(self, woblink_id, doc):
 554         from librarian.cover import make_cover
 555         # TODO Labe
 556         # A5 @ 300ppi.
 557         cover = make_cover(doc.meta, cover_class='m-label', width=1748, height=2480)
 558         content = io.BytesIO()
 559         cover.final_image().save(content, cover.format)
 560         content.seek(0)
 561         self.upload_file(
 562             woblink_id,
 563             doc.meta.url.slug + '.jpeg',
 564             content,
 565             'UploadCover',
 566             'cover',
 567             cover.mime_type()
 568         )