diff --git a/config.example.ini b/config.example.ini index 9c23386..32eaa30 100644 --- a/config.example.ini +++ b/config.example.ini @@ -80,3 +80,14 @@ username = AutoTranslateBot # DeepL configuration deeplendpoint = https://api-free.deepl.com/v2/translate deeplapikey = MySecretDeepLAPIKey + + +[transfer] +# The mediawiki environment from where we copy it, for example 'test' +source_site = test +# The mediawiki environment where we store the content above, for example 'local' +destination_site = local +# The username for the destination site +destination_username = CorrectBot +# Example call for transfer: +# python3 transfer.py "A_Daily_Prayer" "fr" diff --git a/pywikitools/correctbot/bot.py b/pywikitools/correctbot/bot.py index 61a995b..962168a 100644 --- a/pywikitools/correctbot/bot.py +++ b/pywikitools/correctbot/bot.py @@ -25,7 +25,7 @@ def __init__(self, config: ConfigParser, simulate: bool = False): family = Family() self.site = pywikibot.Site(code=code, fam=family, user=self._config.get('correctbot', 'username')) # Set throttle to 0 to speed up write operations (otherwise pywikibot would wait up to 10s after each write) - self.site.throttle.setDelays(delay=0, writedelay=0, absolute=True) + self.site.throttle.set_delays(delay=0, writedelay=0, absolute=True) self.fortraininglib: ForTrainingLib = ForTrainingLib(family.base_url(code, ''), family.scriptpath(code)) diff --git a/pywikitools/test/test_transfer.py b/pywikitools/test/test_transfer.py new file mode 100644 index 0000000..b4ee31a --- /dev/null +++ b/pywikitools/test/test_transfer.py @@ -0,0 +1,54 @@ +from configparser import ConfigParser +import unittest +from unittest.mock import patch +import sys +from transfer import TransferTool +sys.path.append('../../') # Is there a better way to do it? + + +class TestTransferTool(unittest.TestCase): + @patch('pywikibot.Site', autospec=True) + def setUp(self, mock_pywikibot_site): + mock_pywikibot_site.return_value.logged_in.return_value = True + config = ConfigParser() + config.read_dict({"transfer": {"source_site": "test", + "destination_username": "User2", + "destination_site": "local"}}) + self.transfer_tool = TransferTool(config) + + @patch('pywikibot.Page') + def test_upload(self, mock_page): + self.transfer_tool.upload("Test_Page/1/fr", "Test transfer") + mock_page.return_value.save.assert_called_once() + + @patch('pywikibot.Page') + def test_upload_with_message(self, mock_page): + self.transfer_tool.upload("Test_Page/1/fr", "Test transfer", "Test") + mock_page.return_value.save.assert_called_once_with( + summary="Test") + + @patch('pywikibot.Page') + def test_transfer(self, mock_page): + self.transfer_tool.transfer("A_Daily_Prayer", "fr") + mock_page.return_value.save.assert_called() + + @patch('pywikibot.Page') + def test_upload_created(self, mock_page): + mock_page.return_value.exists.return_value = False + self.transfer_tool.upload("Test_Page/2/it", "Test transfer", "v1") + self.assertEqual(self.transfer_tool.created, 1) + self.assertEqual(self.transfer_tool.modified, 0) + self.assertEqual(self.transfer_tool.unchanged, 0) + + @patch('pywikibot.Page') + def test_upload_modified(self, mock_page): + mock_page.return_value.exists.return_value = True + mock_page.return_value.text.return_value = "Old entry" + self.transfer_tool.upload("Test_Page/2/it", "New entry", "v4") + self.assertEqual(self.transfer_tool.created, 0) + self.assertEqual(self.transfer_tool.modified, 1) + self.assertEqual(self.transfer_tool.unchanged, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/transfer.py b/transfer.py new file mode 100644 index 0000000..85d83d0 --- /dev/null +++ b/transfer.py @@ -0,0 +1,98 @@ +import logging +from os.path import abspath, dirname, join +from typing import Optional +import pywikibot +import argparse +from pywikitools.family import Family +from pywikitools.fortraininglib import ForTrainingLib +from configparser import ConfigParser +from pywikitools.lang.translated_page import TranslatedPage + +TIMEOUT: int = 30 # Timeout after 30s (prevent indefinite hanging when there is network issues) + + +class TransferTool: + def __init__(self, config: ConfigParser): + if not config.has_option('transfer', 'source_site') or \ + not config.has_option('transfer', 'destination_site') or \ + not config.has_option('transfer', 'destination_username'): + raise RuntimeError("Missing settings for transfer in config.ini") + + self.logger: logging.Logger = logging.getLogger('pywikitools.transfer') + + self.unchanged: int = 0 + self.modified: int = 0 + self.created: int = 0 + + self.source_site = config.get('transfer', 'source_site') + self.destination_site = config.get('transfer', 'destination_site') + + family = Family() + + self.destination_wiki_site = pywikibot.Site(code=self.destination_site, fam=family, + user=config.get('transfer', 'destination_username')) + if not self.destination_wiki_site.logged_in(): + self.destination_wiki_site.login() + if not self.destination_wiki_site.logged_in(): + raise RuntimeError("Login with pywikibot failed to destination site failed.") + # Set throttle to 0 to speed up write operations (otherwise pywikibot would wait up to 10s after each write) + self.destination_wiki_site.throttle.set_delays(delay=0, writedelay=0, absolute=True) + + self.source_fortraininglib: ForTrainingLib = ForTrainingLib(family.base_url(self.source_site, ''), + family.scriptpath(self.source_site)) + self.destination_fortraininglib: ForTrainingLib = ForTrainingLib(family.base_url(self.destination_site, ''), + family.scriptpath(self.destination_site)) + + def transfer(self, page_name, language_code, message=None): + source_translation_page: Optional[TranslatedPage] = self.source_fortraininglib.get_translation_units( + page_name, language_code) + + if source_translation_page is None: + raise RuntimeError("Could not get translation units from source site") + + self.unchanged: int = 0 + self.modified: int = 0 + self.created: int = 0 + + for source_translation_unit in source_translation_page: + source_translation = source_translation_unit.get_translation() + self.upload(f"{source_translation_unit.identifier}/{language_code}", + source_translation, message) + + numTotal = self.unchanged + self.modified + self.created + print(f"Transfer of {numTotal} elements for '{page_name}/{language_code}' " + + "from '{self.source_site}' to '{self.destination_site}' completed.") + print(f"unchanged: {self.unchanged} | modified: {self.modified} | created: {self.created}") + + def upload(self, identifier: str, translated_text: str, message=None): + """Transfer a worksheet from one mediawiki system to another one""" + destination_mediawiki_page = pywikibot.Page(self.destination_wiki_site, f"Translations:{identifier}") + + if not destination_mediawiki_page.exists(): + self.created += 1 + elif destination_mediawiki_page.text == translated_text: + self.unchanged += 1 + else: + self.modified += 1 + + destination_mediawiki_page.text = translated_text + + if message is None: + destination_mediawiki_page.save() + else: + destination_mediawiki_page.save(summary=message) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description="Transfer a worksheet for a certain language from one site to another site.") + parser.add_argument("worksheet_name", help="Name of the worksheet to transfer.") + parser.add_argument("language_code", help="Target language code for transfer.") + parser.add_argument("--message", nargs=1, help="This is a test") + args = parser.parse_args() + + config = ConfigParser() + config.read(join(dirname(abspath(__file__)), "config.ini")) + + transfer_tool = TransferTool(config) + transfer_tool.transfer(args.worksheet_name, args.language_code, args.message)