【Python】自作のサンプルコード集

Python の単体コード集
大半は、Ubuntu 上の Python3 により, 作成するなどした比較的小さなコード集。

1章 文字列操作
 1.1 パターンマッチによる抽出
   re1.py
   re2.py

2章 時刻操作
 2.1 現在時刻を取得する
   now1.py

3章 sqlite3 との連携
 3.1 DB・TBLの作成〜データ登録
   CollectSummary.py

4章 pexpect との連携
 4.1 基本的な操作メモ
   pexpect1.py
   pexpect2.py
   pexpect3.py
   pexpect4.py

5章 getopt の操作
 5.1 オプション「ho:v」をサポートする
   getopt1.py

6章 GitPython を使った Git の操作
 6.1 git clone, log, checkout をする
   gitpython1.py

 
 

1.1

 
re1.py
入力ファイルを読み込み, 特定のパターンに一致した場合に行番号とその内容を表示する。
ここでは sample.txt を読み込む。
 

re2.py
入力ファイルを読み込み, 特定のパターンに一致した場合に一致箇所のみを抽出する。
 

re1.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import  os
import  sys
import  re

if not os.path.isfile(sys.argv[1]):
  exit

with open(sys.argv[1]) as f:
  pattern = r"\[(.*)\]\[(.*)\]\[(.*)\] (.*)"
  for i, line in enumerate(f, 0):
    m = re.match(pattern, line)
    if m:
      print(line, end="")

 

re2.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import  os
import  sys
import  re

if not os.path.isfile(sys.argv[1]):
  exit

with open(sys.argv[1]) as f:
  pattern = r"\[(.*)\]\[(.*)\] (.*)"
  for i, line in enumerate(f, 0):
    m = re.match(pattern, line)
    if m:
      print('---------------------')
      print(m.group(1))
      print(m.group(2))
      print(m.group(3))

 
sample.txt

[AppX][iOS11] Next application for programming.
[Tips][Windows10] Tips of Windows10.

 
 

2.1

 
時刻関数の使用方法。

now1.py
現在時刻を取得する
 

now1.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import  os
import  sys
import  re
import  datetime

now = datetime.datetime.now()
now.strftime("%Y/%m/%d %H:%M:%s")
print(now)

str = "ただいまの時刻は %s です。" % now
print(now)

 
 

3.1

 
CollectSummary.py
ログファイル(ここでは summary.log) に書かれた内容からパターンに一致した情報を SQLite3 の DB に追加登録する。
作成したい DB ファイルを第1引数に指定することで書き出される。
(例) ./CollectSummary.py Result.db として実行すると Result.db が生成される。

なお、本プログラムでは DETAILS テーブルを作成しているものの使用していない。
 

CollectSummary.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from    collections     import  OrderedDict
import  sys
import  sqlite3
import  re
import  datetime

class   CollectSummary:

  def __init__(self, *args):
    # DB ファイル名
    self.m_database = args[0]
    # DB ファイルが無ければ作る
    self.m_func_create_db()
    # DB ファイル内にテーブルを作る
    self.m_func_create_tbl()
    # 現在時刻を取得する
    self.m_now = datetime.datetime.now().strftime("%Y/%m/%d %H:%M")
    pass

  def m_func_create_db(self):
    # DB に接続する
    self.m_conn = sqlite3.connect(self.m_database)

  def m_func_create_tbl(self):
    # SUMMARY テーブルがあれば作らない
    CMDs    =  { "SUMMARY" : """
                  CREATE TABLE IF NOT EXISTS SUMMARY  
                  (
                  date    verchar(20),
                  status  verchar(10),
                  project verchar(10),
                  title   verchar(20) 
                  )
                 """
          # 本コードでは使わないが、DETAILS テーブル作成時のコマンド
           "DETAILS" : """
                  CREATE TABLE IF NOT EXISTS DETAILS
                  (
                  inspect     verchar(20),
                  machine     verchar(20),
                  exit_code   integer,
                  reason      verchar(20)
                  )
                 """
           }
    # SUMMARY, DETAILS 2つのテーブルを作成する
    for key in CMDs:
      print("[DEBUG]: {0}".format(key))
      self.m_conn.execute(CMDs[key])

  def m_func_save_to_db(self, **kwargs):
    for key in kwargs:
      if key == "SUMMARY":
#               print("{0} => {1}".format(key, kwargs[key]))
        with open(kwargs[key]) as f:
          pattern = r"\[(.*)\]\[(.*)\]\[(.*)\] (.*)"
          for _, line in enumerate(f, 0):
            m = re.match(pattern, line)
            if m:
              sql_cmd = 'INSERT INTO SUMMARY (date, status, project, title) VALUES( ?, ?, ?, ? )'
              values = ( self.m_now, m.group(1), m.group(2), m.group(4))
              self.m_conn.execute(sql_cmd, values)
              print(sql_cmd, values)
      else:
        pass    # Not Support

  def m_func_close_db(self):
    self.m_conn.commit()
    self.m_conn.close()

if __name__ == '__main__':
  
  cs      = CollectSummary(sys.argv[1])
  infiles = { 'SUMMARY' : 'summary.log' }
  cs.m_func_save_to_db( **infiles )
  cs.m_func_close_db()

 
summary.log

[Error][Linux][HP Desktop] Programming machine
[Warn][Win10][Dell] Office work machine
[Critila][Linux][ASUS] File server

 
 

4.1

 
以下のコードの大半は、公式ページを流用して作成した。
なお, pexpect のインストール方法は次の通り。

Pexpect のインストール

  • 本家サイト

https://pexpect.readthedocs.io/en/stable/

  • インストール
% pip3 install pexpect
 pip3 install pexpect
Collecting pexpect
  Downloading https://files.pythonhosted.org/packages/b9/3c/e51a76fe78a877afd9acd95a51e3b7610c0983b19ef5a5dfa11735a619b8/pexpect-4.5.0-py2.py3-none-any.whl (57kB)
    100% |████████████████████████████████| 61kB 2.6MB/s 
Collecting ptyprocess>=0.5 (from pexpect)
  Downloading https://files.pythonhosted.org/packages/ff/4e/fa4a73ccfefe2b37d7b6898329e7dbcd1ac846ba3a3b26b294a78a3eb997/ptyprocess-0.5.2-py2.py3-none-any.whl
Installing collected packages: ptyprocess, pexpect
Successfully installed pexpect-4.5.0 ptyprocess-0.5.2

 

pexpect1.py
spawn せずに ls -la を呼び出すのみ
 

pexpect2.py
spawn を使ったパターン。spawn の型は以下である。

def __init__(self, command, args=[], timeout=30, maxread=2000,
             searchwindowsize=None, logfile=None, cwd=None, env=None,
             ignore_sighup=False, echo=True, preexec_fn=None,
             encoding=None, codec_errors='strict', dimensions=None,
             use_poll=False):
https://pexpect.readthedocs.io/en/stable/_modules/pexpect/pty_spawn.html#spawn

 

pexpect3.py
spawn して lsコマンドを呼び出し、その結果をファイルに書き出す。
なお、/bin/bash -c を使わずに、spawn('ls -la /var > /tmp/log.txt") としてもリダイレクトはできない。「*」「|」 も同様とのこと。
 

pexpect4.py

1. cat によって標準入力(stdin)を監視する
2. 入力された値が指定したパターンに一致するかチェックする
3. 一致したら所定の処理をする (ファイルにも処理内容をダンプする)

 

参考にしたサイト

 

 

pexpect1.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*- 
import  sys
import  pexpect

if __name__ == '__main__':
  pexpect.run('ls -la')  # ' ls -la ' はエラーになるので注意すること

 

pexpect2.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*- 
import  sys
import  pexpect

if __name__ == '__main__':
  # OK パターン
  try:
    child = pexpect.spawn('ls -la /var')
  except  Exception as e:
    print("Exception[{0}] was occured".format(str(e)))
  else:
    print("SUCCESS: 'ls -la /var'");
  finally:
    child.expect(pexpect.EOF)

#   # spawn の第2引数は args=[] なので、下記はエラーになる
#   try:
#       child = pexpect.spawn('ls', '-la', '/var')
#   except  Exception as e:
#       print("Exception[{0}] was occured".format(str(e)))
#   else:
#       print("SUCCESS: 'ls',  '-la',  '/var'");
#   finally:
#       child.expect(pexpect.EOF)

  # OK パターン
  try:
    child = pexpect.spawn('ls', ['-la', '/var'])
  except  Exception as e:
    print("Exception[{0}] was occured".format(str(e)))
  else:
    print("SUCCESS: 'ls', ['-la', '/var']")
  finally:
    child.expect(pexpect.EOF)

 

pexpect3.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*- 
import  sys
import  pexpect

if __name__ == '__main__':
  child = pexpect.spawn('/bin/bash -c "ls -la /var > /tmp/log.txt"')
  child.expect(pexpect.EOF)

 

pexpect4.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*- 
import  sys
import  pexpect

if __name__ == '__main__':

  child = pexpect.spawn('/bin/cat')

  f = open('/tmp/log.txt2', 'wb') # 'w' is invalid
  child.logfile = f

  child.send("123456789\n")
  child.expect(r"(345)..(.)")     # match against terminal echo

  print("Match: child.before=>{0}".format(child.before))     #=> Match: child.before=>b'12'
  print("Match: child.after=>{0}".format(child.after))           #=> Match: child.after=>b'345678'
  print("child.match.groups=>{0}".format(child.match.groups()))    #=> child.match.groups=>(b'345', b'8')

  child.terminate()               # end of expect mode
  child.expect(pexpect.EOF)

 
 

公式ドキュメント

getopt — C-style parser for command line options — Python 3.7.1 documentation
 
getopt_1.py
オプション書式 "ho:v" をサポートするように実装する場合のコード例。
 

getopt1.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*- 
import  getopt
import  sys

OPTION_FORMAT="ho:v"    # サポートするオプション

def usage():
  sys.stderr.write('Usage: {0} "{1}"'.format(sys.argv[0], OPTION_FORMAT))

def main():
  try:
    opts, args = getopt.getopt(sys.argv[1:], OPTION_FORMAT, ["help", "output="])
  except getopt.GetoptError as err:
    # サポート外のオプション指定時はここに進む
    # オプション形式間違い ( -o の後に引数なしの場合など)
    print("0: {0}".format(sys.argv[0]))
    print(err)
    usage()
    sys.exit(2)

  # 引数無しの場合はここに進む
  output = None
  verbose = False

  for o, a in opts:
    if o == "-v":
      verbose = True
    elif o in ("-h", "--help"):
      usage()
      sys.exit(0)
    elif o in ("-o", "--output"):
      output = a
    else:
      assert False, "unhandled option"

  if verbose == True:
    print("verbose mode")
  if output:
    print("output mode")
  if verbose != True and not output:
    print("option is nothing")

  print(sys.argv) # 実行ファイルと引数をダンプする

if __name__ == '__main__':

  main()

 
 

6.1

 
GitPython を使ったサンプルコード集。

GitPython のインストール

  • 本家サイト

 https://github.com/gitpython-developers/GitPython

  • GitPython のインストール
    • 使用する Python は 3.5.3 なので pip3 install GitPython でインストールする
    • もしも Python が 2.X 系なら pip install GitPython とすること
% pip3 install GitPython
Collecting GitPython
  Downloading https://files.pythonhosted.org/packages/88/9c/b462dddb492204417f88d538b0931e87631f2a98afe89842929f4ed9ca5b/GitPython-2.1.9-py2.py3-none-any.whl (447kB)
    100% |████████████████████████████████| 450kB 1.3MB/s 
Collecting gitdb2>=2.0.0 (from GitPython)
  Downloading https://files.pythonhosted.org/packages/e0/95/c772c13b7c5740ec1a0924250e6defbf5dfdaee76a50d1c47f9c51f1cabb/gitdb2-2.0.3-py2.py3-none-any.whl (63kB)
    100% |████████████████████████████████| 71kB 813kB/s 
Collecting smmap2>=2.0.0 (from gitdb2>=2.0.0->GitPython)
  Downloading https://files.pythonhosted.org/packages/e3/59/4e22f692e65f5f9271252a8e63f04ce4ad561d4e06192478ee48dfac9611/smmap2-2.0.3-py2.py3-none-any.whl
Installing collected packages: smmap2, gitdb2, GitPython
Successfully installed GitPython-2.1.9 gitdb2-2.0.3 smmap2-2.0.3

 

gitpython1.py

本コードでは以下の Git コマンドを実行している。

% git clone
% git log
% git checkout

 
より具体的な処理内容は以下の通り。ただし、rm -rf 処理は危険なのでコメントアウトしている。

% rm -rf /tmp/repo
% mkdir /tmp/repo
% git clone -b master https://github.com/Shougo/echodoc.vim
% cd echodoc
% git log -5
% git checkout -b new_branch

 

参考にしたページ
GitPythonを使う - Qiita
python - How to check out a branch with GitPython - Stack Overflow
http://gitpython.readthedocs.io/en/stable/tutorial.html#examining-references
GitのログをGitPythonを使って集計しOrangeを使ってアソシエーション分析する - Qiita

 

gitpython1.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import git  # https://github.com/gitpython-developers/GitPython
import traceback
import shutil
from datetime import datetime as dt 
from time import *
from collections import OrderedDict

WORKDIR = '/tmp/'
TARGET = 'https://github.com/Shougo/echodoc.vim'

def do_rmdir( *args ) -> "rm -rf をするが、間違えると危険なので要注意" :
  for p in args:
    try:
      # shutil.rmtree(p)
      print("[Disable] rm -rf {0}".format(p))
    except:
      sys.exit(1)
    else:
      pass

def do_mkdir( *args ):
  for p in args:
    try:
      if not os.path.isdir(p):
        print("mkdir {0}".format(p))
        os.makedirs(p)
      else:
        print("{0} is alreadly existing".format(p))
    except:
      sys.exit(1)
    else:
      pass

if __name__ == '__main__':

  old_cwd = os.getcwd()
  od = OrderedDict()

  try:
    _local_repo = os.path.join(WORKDIR, 'repo')
    _branch = 'master'
    _log_num = 5
    _new_branch= 'new_branch'
    do_rmdir( _local_repo )  # rm -rf /tmp/repo
    do_mkdir( _local_repo )  # mkdir /tmp/repo

    # git clone
    print("[DEBUG] git clone -b {0} {1}".format(_branch, TARGET))
    repo = git.Repo.clone_from( TARGET, _local_repo, branch=_branch)

    # git log -5 を整形して表示する
    print("[DEBUG] git log -{0}".format(_log_num))
    for item in repo.iter_commits(_branch, max_count=_log_num):
      d = dt.fromtimestamp(item.authored_date).strftime("%Y-%m-%d %H:%M:%S")
      h = item.hexsha
      a = item.author
      od[h] = { d : a }
      print("[{0}] {1} [by {2}]".format(d, h, a))

    # Ordereddict の結果
    print("---- Ordereddict の結果 ----")
    for h,v in od.items():
      for k,a in v.items():
        print("[{0}] {1} [by {2}]".format(k, h, a))

    # git checkout -b new_branch をする場合
    print("[DEBUG] git checkout -b {0}".format(_new_branch))
    repo.git.checkout('-b', _new_branch)

  except:
    e = traceback.print_exc()
    print('エラー原因: {0}'.format(e))
    pass
  else:
    pass
  finally:
    os.chdir(old_cwd)