wuxiaworld.com on LNCrawler v2.29.4

[budikesuma]

@dipu-bd
Hello, dev,

I would like some help implementing the correct script codes to scrape the bottom of the chapters of wuxiaworld novels containing Translator's Thoughts.

The script codes only work on LNCrawler version 2.29.4 on pyDroid3 (it might work on PC/Laptop too, I haven`t tried it yet).

As far as my search on wuxiaworld.com, it works pretty much perfectly on almost all WW novels, except for Archfiend and The Grandmaster Strategist (don`t know if there are html source code differences between WW novels).

With pyDroid3, I managed to download 500 chapters/in 36 seconds, I haven`t tried 1000+ chapters yet.

I disabled the login option part of the script code since it conflicts with other scripts (maybe you can enable the login option capability again?).

You should make sure to uninstall any other version of LNCrawler first, after that you should make sure the 4 versions of libraries below:

idna==2.10
chardet==3.0.4
urllib3==1.24.3
grpcio==1.40.0 (make sure to uninstall pyease-grpc first, most important!!!)

If the version of the above libraries is different from the one installed on your device, make sure to uninstall it first.

I will post the script codes after this for you to check and fix.
Thank you in advance...

[budikesuma]

Using this novel link, since it contains many Translator's Thoughts and footnotes(clickable) to try.

URL Link: https://www.wuxiaworld.com/novel/a-will-eternal

.

.
.

wuxiacom.proto

syntax = "proto3";

package wuxiaworld.api.v2;

/*
 * Generator:
 $ pip install protobuf grpcio grpcio-tools
 $ python -m grpc_tools.protoc -I . --python_out=. wuxiacom.proto
 */

message StringValue {
    string value = 1;
}
message BoolValue {
    bool value = 1;
}
message Int32Value {
    int32 value = 1;
}

message NovelChapterInfo {
    ChapterItem firstChapter = 1;
    ChapterItem latestChapter = 2;
    Int32Value chapterCount = 3;
    repeated ChapterGroupItem chapterGroups = 4;
}

message NovelItem {
    int32 id  = 1;
    string name = 2;
    StringValue coverUrl = 10;
    StringValue authorName = 13;
    NovelChapterInfo chapterInfo = 23;
}

message ChapterItem {
    int32 entityId = 1;
    string name = 2;
    string slug = 3;
    StringValue content = 5;
    int32 novelId = 6;
    StringValue translatorThoughts = 19;
}

message ChapterGroupItem {
    int32 id = 1;
    string title = 2;
    int32 order = 3;
    repeated ChapterItem chapterList = 6;
}

message GetChapterByProperty {
    oneof byProperty {
        int32 chapterId = 1;
        ByNovelAndChapterSlug slugs = 2;
    }

    message ByNovelAndChapterSlug {
        string novelSlug = 1;
        string chapterSlug = 2;
    }
}

message GetNovelRequest {
    oneof selector {
        int32 id = 1;
        string slug = 2;
    }
}

message GetNovelResponse {
    NovelItem item = 1;
}

message GetChapterListRequest {
    int32 novelId = 1;
    FilterChapters filter = 2;

    message FilterChapters {
        Int32Value chapterGroupId = 1;
        BoolValue isAdvanceChapter = 2;
    }
}

message GetChapterListResponse {
    repeated ChapterGroupItem items = 1;
}

message GetChapterRequest {
    GetChapterByProperty chapterProperty = 1;
}

message GetChapterResponse {
    ChapterItem item = 1;
}

message UnlockedItem {
    oneof id {
        int32 novelId = 2;
        int32 chapterId = 3;
    }
}

enum UnlockItemMethod {
    UnlockMethodNone = 0;
    UnlockMethodKarma = 1;
    UnlockMethodVip = 2;
    UnlockMethodSponsor = 3;
}

message UnlockItemRequest {
    UnlockItemMethod unlockMethod = 1;
    UnlockedItem item = 2;
}

message UnlockItemResponse {
    UnlockedItem unlockedItem = 1;
}

service Novels {
    rpc GetNovel(GetNovelRequest) returns (GetNovelResponse);
}

service Chapters {
    rpc GetChapterList(GetChapterListRequest) returns (GetChapterListResponse);
    rpc GetChapter(GetChapterRequest) returns (GetChapterResponse);
}

service Unlocks {
    rpc UnlockItem(UnlockItemRequest) returns (UnlockItemResponse);
}

[budikesuma]

wuxiacom.py

# -*- coding: utf-8 -*-
import logging
import re

from lncrawl.core.crawler import Crawler

logger = logging.getLogger(__name__)

try:
    from google.protobuf.json_format import MessageToDict
    from google.protobuf.message import Message

    from lncrawl.utils.sonora.client import insecure_web_channel
    from lncrawl.etc import wuxiacom_pb2 as proto
except:
    pass

class WuxiaComCrawler(Crawler):
    base_url = ['https://www.wuxiaworld.com/']

    def initialize(self):
        self.home_url = 'https://www.wuxiaworld.com'
        self.grpc = insecure_web_channel(f"https://api2.wuxiaworld.com")
    # end def

##login
##    def login(self, email: str, password: str) -> None:
##        self.bearer_token = email + ' ' + password

    def read_novel_info(self):
        slug = re.findall(r'/novel/([^/]+)', self.novel_url)[0]
        logger.debug('Novel slug: %s', slug)

        client = self.grpc.unary_unary(
            '/wuxiaworld.api.v2.Novels/GetNovel',
            request_serializer=proto.GetNovelRequest.SerializeToString,
            response_deserializer=proto.GetNovelResponse.FromString,

##headers={ 'authorization': self.bearer_token, }
        )
        response = client(proto.GetNovelRequest(slug=slug))
        assert isinstance(response, Message)
        novel = MessageToDict(response)['item']

        self.novel_title = novel['name']
        logger.info('Novel title = %s', self.novel_title)
        
        self.novel_cover = novel['coverUrl']['value']
        logger.info('Novel cover = %s', self.novel_cover)

        self.novel_author = novel['authorName']['value']
        logger.info('Novel author = %s', self.novel_author)

        client = self.grpc.unary_unary(
            '/wuxiaworld.api.v2.Chapters/GetChapterList',
            request_serializer=proto.GetChapterListRequest.SerializeToString,
            response_deserializer=proto.GetChapterListResponse.FromString,

##headers={ 'authorization': self.bearer_token, }
        )
        response = client(proto.GetChapterListRequest(novelId=novel['id']))
        volumes = MessageToDict(response)['items']
        
        for group in sorted(volumes, key=lambda x: x.get('order', 0)):
            vol_id = len(self.volumes) + 1
            self.volumes.append({
                'id': vol_id,
                'title': group['title'],
            })
            for chap in group['chapterList']:
                chap_id = len(self.chapters) + 1
                self.chapters.append({
                    'id': chap_id,
                    'volume': vol_id,
                    'title': chap['name'],
                    'entityId': chap['entityId'],
                    'url': 'https://www.wuxiaworld.com/novel/%s/%s' % (slug, chap['slug']),
                })
            # end for
        # end for
    # end def

    def download_chapter_body(self, chapter):
        client = self.grpc.unary_unary(
            '/wuxiaworld.api.v2.Chapters/GetChapter',
            request_serializer=proto.GetChapterRequest.SerializeToString,
            response_deserializer=proto.GetChapterResponse.FromString,

##headers={ 'authorization': self.bearer_token, }
        )
        property = proto.GetChapterByProperty(chapterId=chapter['entityId'])
        response = client(proto.GetChapterRequest(chapterProperty=property))

        chapter = MessageToDict(response)['item']['content']['value']
        
#        soup = self.make_soup('<main>' + chapter['content']['value'] + '</main>')
#        body = soup.find('main')
#        chapter = str(body)

        if 'translatorThoughts' in MessageToDict(response)['item']:
            chapter += '<hr/>'
            chapter += "<blockquote><b>Translator's Thoughts</b>"
            chapter += MessageToDict(response)['item']['translatorThoughts']
            chapter += "</blockquote>"

        chapter = re.sub(r'(background-)?color: [^>";]+[; ]*', '', chapter)  ##Fixed: Random color issue in chapters content##
        
        return chapter

#        self.clean_contents(body)
#        return str(body)
    # end def
# end class

[budikesuma]

crawler.py

# -*- coding: utf-8 -*-
"""
Crawler application
"""
import itertools
import logging
import random
import re
import ssl
import sys
import unicodedata
from abc import ABC, abstractmethod
from concurrent.futures import ThreadPoolExecutor
from threading import Semaphore
from typing import Dict, List
from urllib.parse import urlparse

import cloudscraper
from bs4 import BeautifulSoup
from bs4.element import Comment, Tag
from requests import Response, Session

from ..assets.user_agents import user_agents
from ..utils.cleaner import TextCleaner
from ..utils.ssl_no_verify import no_ssl_verification
from .exeptions import LNException

logger = logging.getLogger(__name__)


LINE_SEP = '<br>'

INVISIBLE_CHARS = [c for c in range(sys.maxunicode) if unicodedata.category(chr(c)) in {'Cf', 'Cc'}]
NONPRINTABLE = itertools.chain(range(0x00, 0x20), range(0x7f, 0xa0), INVISIBLE_CHARS)
NONPRINTABLE_MAPPING = {character: None for character in NONPRINTABLE}

MAX_CONCURRENT_REQUEST_PER_DOMAIN = 15
REQUEST_SEMAPHORES: Dict[str, Semaphore] = {}

def get_domain_semaphore(url):
    host = urlparse(url).hostname or url
    if host not in REQUEST_SEMAPHORES:
        REQUEST_SEMAPHORES[host] = Semaphore(MAX_CONCURRENT_REQUEST_PER_DOMAIN)
    return REQUEST_SEMAPHORES[host]


class Crawler(ABC):
    '''Blueprint for creating new crawlers'''

    def __init__(self) -> None:
        self._destroyed = False
        self.executor = ThreadPoolExecutor(max_workers=5)
        self.cleaner = TextCleaner()

        # Initialize cloudscrapper
        try:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            self.scraper = cloudscraper.create_scraper(
                # debug=True,
                ssl_context=ctx,
                browser={
                    'custom': random.choice(user_agents),
                    #'browser': 'chrome',
                    #'platform': 'windows',
                    #'mobile': False
                }
            )
        except Exception as err:
            logger.exception('Failed to initialize cloudscraper')
            self.scraper = Session()
        # end try

        # Must resolve these fields inside `read_novel_info`
        self.novel_title = ''
        self.novel_author = ''
        self.novel_cover = None
        self.is_rtl = False

        # Each item must contain these keys:
        # `id` - 1 based index of the volume
        # `title` - the volume title (can be ignored)
        self.volumes = []

        # Each item must contain these keys:
        # `id` - 1 based index of the chapter
        # `title` - the title name
        # `volume` - the volume id of this chapter
        # `volume_title` - the volume title (can be ignored)
        # `url` - the link where to download the chapter
        self.chapters = []

        # Other stuffs - not necessary to resolve from crawler instance.
        self.home_url = ''
        self.novel_url = ''
        self.last_visited_url = None
    # end def

    # ------------------------------------------------------------------------- #
    # Implement these methods
    # ------------------------------------------------------------------------- #

    def initialize(self) -> None:
        pass
    # end def

    def login(self, email: str, password: str) -> None:
        pass
    # end def

    def logout(self) -> None:
        pass
    # end def

    def search_novel(self, query) -> List[Dict[str, str]]:
        '''Gets a list of results matching the given query'''
        return []
    # end def

    @abstractmethod
    def read_novel_info(self) -> None:
        '''Get novel title, autor, cover etc'''
        raise NotImplementedError()
    # end def

    @abstractmethod
    def download_chapter_body(self, chapter) -> str:
        '''Download body of a single chapter and return as clean html format.'''
        raise NotImplementedError()
    # end def

    def download_image(self, url) -> bytes:
        '''Download image from url'''
        logger.info('Downloading image: ' + url)
        response = self.get_response(url, headers={
            'accept': 'image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=1'
        })
        return response.content
    # end def

    def get_chapter_index_of(self, url) -> int:
        '''Return the index of chapter by given url or 0'''
        url = (url or '').strip().strip('/')
        for chapter in self.chapters:
            if chapter['url'] == url:
                return chapter['id']
            # end if
        # end for
        return 0
    # end def

    # ------------------------------------------------------------------------- #
    # Helper methods to be used
    # ------------------------------------------------------------------------- #

    def destroy(self) -> None:
        self._destroyed = True
        self.volumes.clear()
        self.chapters.clear()
        self.scraper.close()
        self.executor.shutdown(False)
    # end def

    @property
    def headers(self) -> dict:
        return dict(self.scraper.headers)
    # end def

    def set_header(self, key: str, value: str) -> None:
        self.scraper.headers[key.lower()] = value
    # end def

    @property
    def cookies(self) -> dict:
        return {x.name: x.value for x in self.scraper.cookies}
    # end def
    
    def set_cookie(self, name: str, value: str) -> None:
        self.scraper.cookies[name] = value
    # end def

    def absolute_url(self, url, page_url=None) -> str:
        url = (url or '').strip()
        if len(url) > 1000 or url.startswith('data:'):
            return url
        # end if
        if not page_url:
            page_url = self.last_visited_url
        # end if
        if not url or len(url) == 0:
            return url
        elif url.startswith('//'):
            return self.home_url.split(':')[0] + ':' + url
        elif url.find('//') >= 0:
            return url
        elif url.startswith('/'):
            return self.home_url.strip('/') + url
        elif page_url:
            return page_url.strip('/') + '/' + url
        else:
            return self.home_url + url
        # end if
    # end def

    def is_relative_url(self, url) -> bool:
        page = urlparse(self.novel_url)
        url = urlparse(url)
        return (page.hostname == url.hostname
                and url.path.startswith(page.path))
    # end def

    def __process_response(self, response: Response) -> Response:
        # if response.status_code == 403 and response.reason == 'Forbidden':
        #     raise LNException('403 Forbidden! Could not bypass the cloudflare protection.\n'
        #                     '  If you are running from your own computer, visit the link on your browser and try again later.\n'
        #                     '  Sometimes, using `http` instead of `https` link may work.')

        response.raise_for_status()
        response.encoding = 'utf8'
        self.cookies.update({
            x.name: x.value
            for x in response.cookies
        })
        return response

    def get_response(self, url, **kargs) -> Response:
        if self._destroyed:
            raise LNException('Instance is detroyed')
        # end if

        kargs = kargs or dict()
        #kargs.setdefault('verify', False)
        #kargs.setdefault('allow_redirects', True)
        kargs.setdefault('timeout', (7, 301))  # in seconds
        headers = kargs.setdefault('headers', {})
        headers = {k.lower(): v for k, v in headers.items()}
        #headers.setdefault('user-agent', random.choice(user_agents))
        headers.setdefault('origin', self.home_url.strip('/'))
        headers.setdefault('referer', self.novel_url.strip('/'))

        with get_domain_semaphore(url):
            with no_ssl_verification():
                response = self.scraper.get(url, **kargs)

        self.last_visited_url = url.strip('/')
        return self.__process_response(response)
    # end def

    def post_response(self, url, data={}, headers={}) -> Response:
        if self._destroyed:
            raise LNException('Instance is detroyed')
        # end if

        headers = {k.lower(): v for k, v in headers.items()}
        #headers.setdefault('user-agent', random.choice(user_agents))
        headers.setdefault('content-type', 'application/json')
        headers.setdefault('origin', self.home_url.strip('/'))
        headers.setdefault('referer', self.novel_url.strip('/'))
        logger.debug('POST url=%s, data=%s, headers=%s', url, data, headers)

        with get_domain_semaphore(url):
            with no_ssl_verification():
                response = self.scraper.post(
                    url,
                    data=data,
                    headers=headers,
                    # verify=False,
                    # allow_redirects=True,
                )

        return self.__process_response(response)
    # end def

    def submit_form(self, url, data={}, multipart=False, headers={}) -> Response:
        '''Submit a form using post request'''
        if self._destroyed:
            raise LNException('Instance is detroyed')
        # end if

        content_type = 'application/x-www-form-urlencoded; charset=UTF-8'
        if multipart:
            content_type = 'multipart/form-data'
        # end if
        headers = {k.lower(): v for k, v in headers.items()}
        headers.setdefault('content-type', content_type)

        return self.post_response(url, data, headers)
    # end def

    def get_soup(self, *args, **kwargs) -> BeautifulSoup:
        parser = kwargs.pop('parser', None)
        response = self.get_response(*args, **kwargs)
        return self.make_soup(response, parser)
    # end def

    def make_soup(self, response, parser=None) -> BeautifulSoup:
        if isinstance(response, Response):
            html = response.content.decode('utf8', 'ignore')
        elif isinstance(response, bytes):
            html = response.decode('utf8', 'ignore')
        elif isinstance(response, str):
            html = str(response)
        else:
            raise LNException('Could not parse response')
        # end if
        soup = BeautifulSoup(html, parser or 'lxml')
        if not soup.find('body'):
            raise ConnectionError('HTML document was not loaded properly')
        # end if
        return soup
    # end def

    def get_json(self, *args, **kwargs) -> dict:
        kwargs = kwargs or dict()
        headers = kwargs.setdefault('headers', {})
        headers = {k.lower(): v for k, v in headers.items()}
        headers.setdefault('accept', 'application/json, text/javascript, */*')
        response = self.get_response(*args, **kwargs)
        return response.json()
    # end def

    def post_soup(self, url, data={}, headers={}, parser='lxml') -> BeautifulSoup:
        response = self.post_response(url, data, headers)
        return self.make_soup(response, parser)
    # end def

    def post_json(self, url, data={}, headers={}) -> dict:
        headers = {k.lower(): v for k, v in headers.items()}
        headers.setdefault('accept', 'application/json, text/javascript, */*')
        response = self.post_response(url, data, headers)
        return response.json()
    # end def

    def download_cover(self, output_file) -> None:
        response = self.get_response(self.novel_cover)
        with open(output_file, 'wb') as f:
            f.write(response.content)
        # end with
    # end def

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

    blacklist_patterns = []
    bad_tags = [
        'noscript', 'script', 'style', 'iframe', 'ins', 'header', 'footer',
        'button', 'input', 'amp-auto-ads', 'pirate', 'figcaption', 'address',
        'tfoot', 'object', 'video', 'audio', 'source', 'nav', 'output', 'select',
        'textarea', 'form', 'map',
    ]
    bad_css = [
        '.code-block', '.adsbygoogle', '.sharedaddy', '.inline-ad-slot', '.ads-middle',
        '.jp-relatedposts', '.ezoic-adpicker-ad', '.ezoic-ad-adaptive', '.ezoic-ad',
        '.cb_p6_patreon_button', 'a[href*="patreon.com"]',
    ]
    p_block_tags = [
        'p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6',
        'main', 'aside', 'article', 'div', 'section',
    ]
    unchanged_tags = [
        'pre', 'canvas', 'img'
    ]
    plain_text_tags = [
        'span', 'a', 'abbr', 'acronym', 'label', 'time',
    ]
    substitutions = {
        '"s': "'s",
        '“s': "'s",
        '”s': "'s",
        '&': '&amp;',
        'u003c': '<',
        'u003e': '>',
        '<': '&lt;',
        '>': '&gt;',
    }

    def clean_text(self, text) -> str:
        text = str(text).strip()
        text = text.translate(NONPRINTABLE_MAPPING)
        for k, v in self.substitutions.items():
            text = text.replace(k, v)
        return text
    # end def

    def clean_contents(self, div):
        if not isinstance(div, Tag):
            return div
        # end if
        if self.bad_css:
            for bad in div.select(','.join(self.bad_css)):
                bad.extract()
            # end if
        # end if
        for tag in div.find_all(True):
            if isinstance(tag, Comment):
                tag.extract()   # Remove comments
            elif tag.name == 'br':
                next_tag = getattr(tag, 'next_sibling')
                if next_tag and getattr(next_tag, 'name') == 'br':
                    tag.extract()
                # end if
            elif tag.name in self.bad_tags:
                tag.extract()   # Remove bad tags
            elif hasattr(tag, 'attrs'):
                tag.attrs = {k: v for k, v in tag.attrs.items() if k == 'src'}
            # end if
        # end for
        div.attrs = {}
        return div
    # end def

    def extract_contents(self, tag) -> str:
        self.clean_contents(tag)
        body = ' '.join(self.__extract_contents(tag))
        return '\n'.join([
            '<p>' + x + '</p>'
            for x in body.split(LINE_SEP)
            if not self.__is_in_blacklist(x.strip())
        ])
    # end def

    def __extract_contents(self, tag) -> list:
        body = []
        for elem in tag.contents:
            if isinstance(elem, Comment):
                continue
            if not elem.name:
                body.append(self.clean_text(elem))
                continue
            if elem.name in self.unchanged_tags:
                body.append(str(elem))
                continue
            if elem.name == 'hr':
                body.append(LINE_SEP)
                # body.append('-' * 8)
                # body.append(LINE_SEP)
                continue
            if elem.name == 'br':
                body.append(LINE_SEP)
                continue
            # if not elem.text.strip():
            #     continue

            is_block = elem.name in self.p_block_tags
            is_plain = elem.name in self.plain_text_tags
            content = ' '.join(self.__extract_contents(elem))

            if is_block:
                body.append(LINE_SEP)
            # end if

            for line in content.split(LINE_SEP):
                line = line.strip()
                if not line:
                    continue
                # end if
                if not (is_plain or is_block):
                    line = '<%s>%s</%s>' % (elem.name, line, elem.name)
                # end if
                body.append(line)
                body.append(LINE_SEP)
            # end if

            if body and body[-1] == LINE_SEP and not is_block:
                body.pop()
            # end if
        # end for

        return [x.strip() for x in body if x.strip()]
    # end def

    def __is_in_blacklist(self, text) -> bool:
        if not text:
            return True
        # end if
        if not self.blacklist_patterns:
            return False
        # end if
        pattern = getattr(self, '__blacklist__', None)
        if not pattern:
            pattern = re.compile('|'.join(['(%s)' % p for p in self.blacklist_patterns]))
            setattr(self, '__blacklist__', pattern)
        # end if
        if pattern and pattern.search(text):
            return True
        return False
    # end def

# end class

[dipu-bd]

It would be better to open a pull request with your code.

[dipu-bd]

If you are not sure, how to do it, here is a mini-tutorial.

• Fork this project
• Clone your fork
• Checkout to dev branch
• Modify the files that you need
• Make a commit and push to your forked repo
• Go to Pull Request section, and create a new one, or paste your repository link here, I will create one.