1
\$\begingroup\$

I have a piece of code that downloads acupuncture data from Wikipedia and consolidates it into an acupoint, meridian and extraordinary meridian dictionary stored as .pkl files in the working directory.

import pandas as pd
import wikipedia as wp
import json
from hanziconv import HanziConv as hconv
from copy import deepcopy
from os.path import isfile
import itertools
import re

class Jingluo():

    def __init__(self):
                
        self.yinyang = ("太陰", "陽明", "少陰", "太陽", "厥陰", "少陽" )
        
        try:  
            self.zhengjing, self.xunjing = self.read_meridians()
            self.qijing = self.read_extraordinary_meridians()
            self.xue = self.read_acupoints()            
        except: 
            print("Downloading data from Wikipedia...")
            self.get_data_from_wikipedia()

            self.zhengjing, self.xunjing = self.read_meridians()
            self.qijing = self.read_extraordinary_meridians()
            self.xue = self.read_acupoints()

    @staticmethod
    def get_data_from_wikipedia():
        "Scrape data from wikipedia and output three dataframes: \
        acupoints, meridians and extraordinary meridians."

        html = wp.page("List_of_acupuncture_points").html()
        df = pd.read_html(html)

        # Set meridian, extra meridian and acupoint
        meridians = df[0][['Code', 'Chinese Name', 'English']]
        extraordinary_meridians = df[1][['Code','Name','Transliteration']]
        extraordinary_meridians.columns = ['ID', 'Name', 'Transliteration']
        extraordinary_meridians = extraordinary_meridians.set_index('ID')

        acupoints = pd.concat(df[2:16])[['Point', 'Name', 'Transliteration']] # standard :16, all :18
        acupoints.columns = ['ID', 'Name', 'Transliteration']
        acupoints = acupoints.set_index('ID')

        # Data cleaning

        ## Fix meridian data
        meridians = deepcopy(meridians)
        meridians['Chinese Name'] = [hconv.toTraditional(item) for item in meridians['Chinese Name']]
        meridians.loc[11]['Code'] = 'LR'

        ## Fix acupoint data
        as_list = acupoints.index.tolist()
        split_list = [item.split('-') for item in as_list]
        tag_list = [tag for tag, sn in split_list]
        sn_list = [sn for tag, sn in split_list]

        tag_list = ["LR" if tag == "Liv" else tag for tag in tag_list]
        tag_list = ["GV" if tag == "Du" else tag for tag in tag_list]
        tag_list = ["CV" if tag == "Ren" else tag for tag in tag_list]
        tag_list = [item.upper() for item in tag_list]

        new_idx_list = [tag + sn for tag, sn in zip(tag_list, sn_list)]
        acupoints.index = new_idx_list

        # Write to disk

        acupoints.to_pickle('acupoints.pkl') # save to pickle file
        meridians.to_pickle('meridians.pkl')
        extraordinary_meridians.to_pickle('extraordinary_meridians.pkl')
        
        print('Done!')
        
    # 穴位
    
    @staticmethod
    def read_acupoints():
        if isfile('acupoints.pkl'):
            ACUPOINTS = pd.read_pickle('acupoints.pkl')
            STRING_LABELS = [name + "(" + index + ")"  for name, index in zip(list(ACUPOINTS['Name']), list(ACUPOINTS.index))]
            # convert to dictionary
            acupoint = {name: {"id": index, "label": name + "(" + index + ")"} for name, index in \
                zip(list(ACUPOINTS['Name']), list(ACUPOINTS.index))}
            
            return acupoint
        
        else:
            print("Acupoints table not found.")

    # 經絡      

    @staticmethod
    def read_meridians():
        if isfile('meridians.pkl'):
            MERIDIANS = pd.read_pickle('meridians.pkl')            
            limb_list = [ re.search("([手足])(.+?[陰陽明])(.+)經", item).group(1)\
                     for item in MERIDIANS['Chinese Name']]
            yinyang_list = [ re.search("([手足])(.+?[陰陽明])(.+)經", item).group(2)\
                        for item in MERIDIANS['Chinese Name']]
            organ_list = [ re.search("([手足])(.+?[陰陽明])(.+)經", item).group(3)\
                      for item in MERIDIANS['Chinese Name']]

            zhengjing = { organ:{"id": index, "name": name, "short_name": organ + "經",\
                      "label": name + "(" + index + ")", "limb": limb, "yinyang": yinyang}\
              for name, index, limb, yinyang, organ in \
             zip(MERIDIANS['Chinese Name'], MERIDIANS['Code'], limb_list, yinyang_list, organ_list)}

            xunjing = (item for item in zhengjing) # generator object; use next(xunjing) to simulate circulation.
        
            return zhengjing, xunjing
        
        else:
            print("Meridians table not found.")

    #奇經
            
    @staticmethod
    def read_extraordinary_meridians():
        if isfile('extraordinary_meridians.pkl'):
            EXTRAORDINARY_MERIDIANS = pd.read_pickle('extraordinary_meridians.pkl') 

            em_name_list = [item.split('; ')[1] for item in EXTRAORDINARY_MERIDIANS['Name']]
            em_name_list = [re.sub('蹺', '蹻', item) for item in em_name_list]

            em_list = [re.sub('脈', '', item) for item in em_name_list]

    #     hand_meridian_list = [ value['臟腑'] for key, value in .items() if '手' in value['肢'] ]

            qijing = {short_hand: {"id": index, "name": name, "label": name + "(" + index + ")"} for short_hand, name, index in \
                    zip(em_list, em_name_list, list(EXTRAORDINARY_MERIDIANS.index))}

            return qijing
        
        else:
            print("Extraordinary Meridians table not found.")

You can view the structure of the dataset by running the above initiation code and calling:

x= Jingluo()

x.zhengjing # meridians
x.qijing # extraordinary meridians
x.xue # acupoint data

This is working fine, but I wonder if there a more efficient and effective way of acquiring and storing the above data?

I would like the dataset to be easily extensible. For example, I might want to tag images to show the position of each acupoint in the xue (i.e. acupoint) dictionary.

It seems to me an SQLite relational database might be more useful for this situation.

\$\endgroup\$
5
  • \$\begingroup\$ Using non-ascii characters for strings and displayable text if ok, but using them for variable names is awful for two reasons: 1- If format is lost, your code may become unrunnable. 2 - If your code will ONLY be read by people who speak your native language, you might write code in that language; but if your code will be read by other people, it should be in english. Specially if the language uses a different character set \$\endgroup\$ Aug 6 '21 at 8:35
  • \$\begingroup\$ Removed Chinese characters in variable names. \$\endgroup\$
    – Sati
    Aug 6 '21 at 9:45
  • 1
    \$\begingroup\$ @m-alorda Overall your point stands, though "awful" is perhaps a stretch, and "format is lost" makes no sense to me. Are you suggesting that UTF-8 is going to cease to exist? \$\endgroup\$
    – Reinderien
    Aug 6 '21 at 13:15
  • \$\begingroup\$ @Reinderien You are right, maybe saying "awful" wasn't the best way to describe it. Regarding format, I may not have explained myself good enough. I'm not saying UTF-8 will cease to exist at all. I've come across code that uses non-ascii characters sush as "ñ" or "á". However, the format was then changed to ascii, and they weren't properly represented (for example "á" was then represented by something like "A~"). I just don't know exactly how it happened, maybe it was the editor that got the settings messed up, or maybe it was after zipping it and/or uploading to a university app \$\endgroup\$ Aug 6 '21 at 14:46
  • \$\begingroup\$ Also, I don't think I deserve to be downvoted just for that. I have actually made an effort to make it clear contextually what the variables mean by way of comments, etc. \$\endgroup\$
    – Sati
    Aug 6 '21 at 16:44
1
\$\begingroup\$

This took me a long time to go through.

First, re. non-ASCII characters for strings - yes(ish), but it's less important to eliminate non-ASCII characters, and more important to make a consistent codebase whose variables are in English for international collaboration purposes. So it's not helping us to replace

class 經絡():

with

class Jingluo():

Rather than only romanisation, you should perform translation, so this should instead be

class Meridian():

When you present your data to the user that's a different story; presentation is where you show localized languages.

wikipedia is not a very helpful library for your purposes. Currently you're using it as a glorified requests, so just use requests instead. More broadly: your current data path is

  • Wikipedia's database, in mediawiki markup; to
  • Wikipedia's renderer outputting HTML; to
  • your Pandas read_html call.

It roughly works but is non-ideal for a list of reasons:

  • HTML is not a data interchange format, and does not represent a stable API.
  • The above trip through a presentation markup is information-lossy. Among (many) other things, you're losing the semantic data about alternate language variants like traditional versus simplified Chinese, and heading titles.

For these reasons you should consider

  • Instead of hitting the Wikipedia website, hit their MediaWiki API with Requests
  • Parse the returned markup with a mediawiki parsing library
  • Pay attention to the heading titles, so that you can more robustly identify where and what your tables are
  • Pay attention to the zh template so that you can explicitly select between simplified and traditional Chinese, something I suspect you care about

Otherwise:

  • Jingluo is full of statics and thus does not deserve to be a class
  • self.yinyang is never used and should be deleted
  • Never bare except. As a bonus: if you were to pay attention to the thrown exception, it's not (!) throwing a FileNotFoundError; but rather is failing to unpack your None return. So
    • do not call isfile;
    • do not else / print;
    • do not except:;
    • let the FileNotFoundError fall through;
    • catch it specifically at the outer level.
  • Don't print('Downloading .... If you really want to see this, send it to a logger instance instead
  • As hinted at above, do not index [0] into your tables sequence; pay attention to the titles in the mediawiki markup. What if a table were inserted between the indices that you currently hard-code?
  • Do not hard-code index 11; instead use a Pandas replacement
  • The moment you're calling tolist, something in your Pandas usage has probably gone wrong. Don't do this. Call into the vectorized string methods.
  • Your split('-') is going to crash for at least two different cases I see in the Wikipedia page. Some names have multiple-dashed-components, and others have no dash separator. One solution is to apply a regex with a digit group and call extractall.
  • Do not bake your disk-writing code into the same method as your cleaning code.
  • You have this weird disconnect between pre-processing and post-processing, where your pre-processed data are in Pandas dataframes and your post-processed data are in dictionaries; and for whatever reason you do the dictionary conversion on every load of the Pickle file. I don't know why this is being done, so in my suggestion there is no post-conversion at all, and some elements of the post-conversion (e.g. parsing out of limb and organ strings) are included in pre-processing.
  • You re-execute your "([手足])(.+?[陰陽明])(.+)經" regex - twice. Don't do this; just use Pandas vectorized extractall.
  • I don't understand what you were trying to accomplish with xunjing = (item for item in zhengjing); making a no-op generator doesn't have any use case that I can think of so you should just return zhengjing itself.
  • re.sub('蹺', '蹻') does not need the regex module and can just be a Pandas-vectorized replace call.

I would like the dataset to be easily extensible. For example, I might want to tag images to show the position of each acupoint in the xue (i.e. acupoint) dictionary.

This is possible in Pandas itself, which can easily store blobs for images. Your data set is very small so I consider SQL overkill.

Suggested

Not exactly equivalent, but that's deliberate based on above

from pprint import pprint
from typing import Tuple, Iterable, Dict, Collection

import pandas as pd
import wikitextparser
from hanziconv import HanziConv as hconv
from os.path import isfile
import re
from requests import get
from wikitextparser import Table, Section, WikiText



# Attempting to translate some terms from
# https://www.ctcmpao.on.ca/resources/forms-and-documents/Standard_Acupuncture_Nomenclature.pdf


Frames = Tuple[
    pd.DataFrame,  # meridians
    pd.DataFrame,  # extraordinary meridians
    Collection[pd.DataFrame],  # acupoints
]

FramesByTitle = Iterable[Tuple[
    str,  # Section title
    pd.DataFrame,  # Dataframe derived from table
]]

FILENAMES = (
    'acupoints.pkl',
    'meridians.pkl',
    'extraordinary_meridians.pkl',
)

COLUMN_MAP = {
    'Code': 'ID',
    'Point': 'ID',
}

NORMALIZED_COLS = ['ID', 'Name', 'Transliteration']


def get_wiki(page_name: str) -> WikiText:
    with get(
        'https://en.wikipedia.org/w/api.php',
        params={
            'action': 'query',
            'prop': 'revisions',
            'rvprop': 'content',
            'rvslots': 'main',
            'format': 'json',
            'titles': page_name,
        },
    ) as resp:
        resp.raise_for_status()
        page, = resp.json()['query']['pages'].values()

    text = page['revisions'][0]['slots']['main']['*']
    return wikitextparser.parse(text)


def templates_to_text(row: Iterable[str]) -> Iterable[str]:
    for cell in row:
        parsed = wikitextparser.parse(cell)

        if len(parsed.templates) > 0:
            template = parsed.templates[0]
            attrs = {arg.name: arg.value for arg in template.arguments}

            if template.name == 'lang':
                # Here you might care to do something special for language metadata
                yield attrs['2']
            elif template.name == 'zh':
                # Here you might want to select between traditional or simplified;
                # in this case we show only simplified which is what you had been
                # doing in em_name_list = [item.split('; ')[1]
                # To show both in Wikipedia style would basically be
                # yield f'{attrs["t"]}; {attrs["s"]}'
                yield attrs['s']
            else:
                yield cell[:template.span[0]]
            continue

        if len(parsed.wikilinks) > 0:
            link = parsed.wikilinks[0]
            yield link.text
            continue

        yield cell


def get_dataframes(doc: WikiText) -> FramesByTitle:
    table_refs: Dict[
        int,  # Starting character index of the table in the document
        Tuple[Section, Table],  # Section and the table it contains
    ] = {}

    # The following hack is needed because wikitextparser includes non-direct-
    # descendant tables in each section

    for section in doc.sections:
        for table in section.tables:
            start, end = table.span
            section_table = table_refs.get(start)
            if section_table is None:
                use = True
            else:
                old_section, old_table = section_table
                use = old_section.level < section.level
            if use:
                table_refs[start] = section, table

    for section, table in table_refs.values():
        columns, *rows = table.data()
        yield section.title, pd.DataFrame(
            [
                tuple(templates_to_text(row))
                for row in rows
            ],
            columns=columns,
        )


def assign_frames(frames: FramesByTitle, standard_only: bool = True) -> Frames:
    """
    Scrape data from wikipedia and output three dataframes:
    acupoints, meridians and extraordinary meridians.
    """

    all_tables = tuple(frames)
    tables: Dict[str, pd.DataFrame] = {
        title: df for title, df in all_tables
        # Excluded due to duplicate title
        if title != 'Extra points'
    }

    if not standard_only:
        deadman, extra = (
            df for title, df in all_tables if title == 'Extra points'
        )
        tables['Extra points (Deadman)'] = deadman
        tables['Extra points'] = extra

    return (
        tables.pop('Twelve Primary Meridians'),
        tables.pop('Eight Extraordinary Meridians'),
        tables.values(),
    )


def clean_meridians(meridians: pd.DataFrame) -> pd.DataFrame:
    meridians.drop(
        meridians.columns.difference(['Code', 'Chinese Name', 'English']),
        axis=1, inplace=True,
    )
    meridians['Chinese Name'] = meridians['Chinese Name'].map(hconv.toTraditional)
    meridians.Code[meridians.Code == 'LV'] = 'LR'

    groups = meridians['Chinese Name'].str.extractall(
        r'(?P<limb>[手足])'
        r'(?P<yinyang>.+?[陰陽明])'
        r'(?P<organ>.+)經'
    )
    groups.reset_index(drop=True, inplace=True)
    meridians = pd.concat((meridians, groups), axis=1)

    return meridians


def normalize(df: pd.DataFrame) -> pd.DataFrame:
    df.rename(columns=COLUMN_MAP, inplace=True)
    df.drop(df.columns.difference(NORMALIZED_COLS), axis=1, inplace=True)
    return df


def clean_extra_meridians(df: pd.DataFrame) -> pd.DataFrame:
    df = normalize(df)
    df.set_index('ID', inplace=True)

    df.Name = (
        df.Name
        .str.replace('蹺', '蹻')
        .str.replace('脈', '')
    )
    return df


def clean_acupoints(all_points: Collection[pd.DataFrame]) -> pd.DataFrame:
    for points in all_points:
        if 'Transliteration' not in points.columns:
            points.rename(columns={'Pinyin': 'Transliteration'}, inplace=True)

    all_points = [normalize(points) for points in all_points]

    acupoints = pd.concat(all_points, ignore_index=True)
    groups = acupoints.ID.str.extractall(
        r'^'             # start
        r'(?P<tag>.*?)'  # tag portion, non-greedy
        r'-?'            # optional separating hyphen
        r'(?P<sn>\d+)'   # serial number portion
        r'$'             # end
    )
    groups.reset_index(drop=True, inplace=True)
    acupoints = pd.concat((acupoints, groups), axis=1)
    acupoints.tag = acupoints.tag.str.upper()
    acupoints.tag.replace(
        {
            'LIV': 'LR',
            'DU': 'GV',
            'REN': 'CV',
        },
        inplace=True,
    )
    acupoints.index = acupoints.tag + acupoints.sn
    return acupoints


def download_or_read() -> Frames:
    try:
        return tuple(pd.read_pickle(fn) for fn in FILENAMES)
    except FileNotFoundError:
        pass

    doc = get_wiki('List_of_acupuncture_points')
    meridians, extras, points = assign_frames(get_dataframes(doc), standard_only=False)

    meridians = clean_meridians(meridians)
    extras = clean_extra_meridians(extras)
    points = clean_acupoints(points)

    write_to_disk(meridians, extras, points)
    return meridians, extras, points


def write_to_disk(*args: pd.DataFrame) -> None:
    for filename, df in zip(FILENAMES, args):
        df.to_pickle(filename)


def main():
    download_or_read()


if __name__ == '__main__':
    main()
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.