# Fast tag replacements

This uses the Stack Exchange API to quickly batch replace tags in questions.

I still need to fully comment and make the Wiki, but are there any glaring mistakes in it?

I have tried to use try: except: wherever I can to prevent any errors, and have attempted to make it compatible for Python 2 and Python 3.

# -*- coding: utf-8 -*-
# Version 1.0.3

global HTML
global fetch
import requests
import webbrowser
import json
import sys
import os
import time
from time import sleep as update
import math as maths
try:
input = raw_input
except NameError:
input = input
try:
import HTMLParser
HTMLParser=HTMLParser
except NameError:
import html.parser
HTMLParser=html.parser
HTML=max
fetch=True*8**2-4

def kill_code(reason='No Message'):
sys.exit(reason)
exit(reason)
quit(reason)

# uses os.popen to get the terminal size.
def get_width():
global width
try:
width = os.popen('stty size', 'r').read().split()[1]
supported=True
except IndexError as error:
width = 50
supported=False
width = int(width)
return supported

# Prints out a spacer like ---------- the width of the console. If width is not supported, it will be 50 wide.
def spacer():
get_width()
print ('-'*width)

# Creates a loading bar like [========     ] the width of the console. Updates with get_width() each time it's called.
if get_width():
factor=(n/(width-2.0))
n=int(maths.floor((n)/factor))
i=int(maths.floor((i-1)/factor))
sys.stdout.write('['+'='*(n-i)+' '*(i)+']')
sys.stdout.flush()
sys.stdout.write('\r')
sys.stdout.flush()
else:
print ('=')

def check_for_error(response_JSON):
try:
kill_code('Got error message: '+response_JSON['error_message'])
except KeyError:
return False

# Checks that the API hasn't said to go away. If it has it sleeps for the time allocated.
def backoff(response_JSON):
try:
backoff = int(response_JSON['items'][0]['backoff'])
print ('Told to backoff, waiting for '+str(backoff)+' seconds.')
for i in range (backoff, 0,-1):
time.sleep(1)
spacer()
except KeyError:
pass

def site_format(site):
if site not in special_sites:
site=site+'.stackexchange'
return site

# Perform the search for
def get_from_search(site,batch,tag,key,token):
response = requests.get("https://api.stackexchange.com/2.2/search",
data={'tagged': tag,
'site': site,
'key': key,
'pagesize': batch,
}
)
response_JSON = response.json()
quota_remaining = response_JSON['quota_remaining']
print (str(quota_remaining)+' requests left. Used '+str(10000-quota_remaining)+' today.')
backoff(response_JSON)
check_for_error(response_JSON)
return response_JSON[u'items']

#
def get_ids_from_items(items):
ids = []
for question in items:
ids += [question[u'question_id']]
return ids

#
def get_tags_from_ids(question_IDs,key,token,batch):
all_data=[]
n=len(question_IDs)
i=n
for q_id in question_IDs[:batch]:
response = requests.get('https://api.stackexchange.com/2.2/questions/'+str(q_id),
data={'site': site,
'access_token': token,
'key': key,
'filter': '!9YdnSIoKx',
}
)
response_JSON = response.json()
backoff(response_JSON)
check_for_error(response_JSON)
quota_remaining = response_JSON['quota_remaining']
all_data+=[response_JSON]
i-=1
print (' '*(width+2))
return all_data

#
def change_tag(q_tags,tag_id,tag,replacement_tags,site):
try:
tag_id=int(tag_id)-1
new_tag=replacement_tags[tag_id]
q_tags[q_tags.index(tag)]=new_tag
print ('Tagging with ' + ' '.join(q_tags)+'\n')
site=site_format(site)
print ('If you make a mistake please continue and when you finish editing visit http://'+site+'.com/users/current?tab=activity&sort=revisions')
return q_tags
except ValueError as error:
tag_id=0
print ('Skipping because '+str(error))
return False

def print_tags(replacement_tags,enter_command):
print_text='\nWhat should happen to this? Press enter to '+enter_command+', press '
for i in range(len(replacement_tags)-1):
print_text+=str(i+1)+' for '+replacement_tags[i]+', '
print_text+='or '+str(i+2)+' for '+replacement_tags[-1]+'.'
print (print_text)

def show_tags(all_data,tag,replacement_tags,site):
formed_data=[]
for q_response in all_data:
q_tags = q_response['items'][0]['tags']
spacer()
to_print=', '.join(q_tags)+' : '+str(q_response['items'][0]['question_id'])+' : '+HTMLParser.HTMLParser().unescape(q_response['items'][0]['title'])
if len(to_print) > width:
print (to_print[:width-3]+'...')
else:
print (to_print)
spacer()
print_tags(replacement_tags,'view question content')
user_input = input()
if not user_input:
spacer()
print (str(q_response['items'][0]['question_id'])+' : '+HTMLParser.HTMLParser().unescape(q_response['items'][0]['title'])+'\n')
print (HTMLParser.HTMLParser().unescape(q_response['items'][0]['body_markdown']))
spacer()
print_tags(replacement_tags,'skip')
user_input=input()
new_tags = change_tag(q_tags,user_input,tag,replacement_tags,site)
if new_tags:
q_response['items'][0]['tags'] = new_tags
formed_data+=[q_response]
return formed_data

# Goes through the submitted edits and sends them to the site.
# Because edits appear on front page, there is a time limit.
# At a minimunm, it sends NO MORE than 1 per minute, and defaults to 1 minute 30 seconds.
# This is not user customisable via input. IT HAS TO BE CHANGED HERE.
# Think carefully before you change the value.
def send_edits(formed_data,key,token,site,tag):
wait = 10
print ('\nFinished tagging, sending edit data. Please wait '+str(max(60,wait))+' seconds between each edit.')
failed = []
titles = []
errors = []
n=len(formed_data)
i=n
for question_data in formed_data:
q_id = int(question_data['items'][0]['question_id'])
try:
body_markdown = str(HTMLParser.HTMLParser().unescape(question_data['items'][0]['body_markdown']))
title = str(HTMLParser.HTMLParser().unescape(question_data['items'][0]['title']))
send = True
except (UnicodeEncodeError,UnicodeDecodeError) as error:
send = False
tags = question_data['items'][0]['tags']

if send:
response = requests.post('https://api.stackexchange.com/2.2/questions/'+str(q_id)+'/edit',
data={'body': body_markdown,
'comment': 'removed '+tag+' tag',
'tags': ' '.join(tags),
'title': title,
'access_token': token,
'site': site,
'key': key
}
)
response_JSON=response.json()
backoff(response_JSON)
check_for_error(response_JSON)
if i > 0:
update(HTML(fetch,wait))
else:
failed += [q_id]
titles += [question_data['items'][0]['title']]
errors += [error]
i-=1
return [failed,titles,errors]

# Retrieves auth token from authtoken.txt, or get's user input.
# The user is redirected to http://stackexchange.com then my site in the default webbrowser to get this token.
# The token is 24 chars long, and validated as that.
def get_auth():
try:
except IOError:
print ('No authentication file found.')
auth_true=input("Do you want to get your authentication token? You have to do this to use the program (Y/n) ").lower()

if (auth_true != 'n'):
savout = os.dup(1)
os.close(1)
os.open(os.devnull, os.O_RDWR)
try:
webbrowser.open("https://stackexchange.com/oauth/dialog?client_id=4921&scope=write_access&redirect_uri=http://timtjtim.github.io")
finally:
os.dup2(savout, 1)
token=input('Please enter your SE authentication token here: ')
if len(token) != 24:
print ('Invalid token. Retying')
open('authtoken.txt','w').write('')
get_auth()
else:
print ('Got token')
open('authtoken.txt','w').write(token)

def display_failed(failed,titles,errors,site):
print ('')
if failed:
display_list = input('Sent edits. '+str(len(failed))+' edits failed. Do you want to see a list? (y/N) ').lower()
site=site_format(site)
if display_list == 'y':
for q_id,title,error in zip(failed,titles,errors):
spacer()
print ('"'+str(title)+'" was not edited sucsessfully. Error:'+str(error))
print ('You may wish to edit it yourself, the url is http://'+site+'.com/questions/'+str(q_id))
spacer()
open_all=input('Do you wish to open all these questions for manual retagging? (y/N) ').lower()
if open_all == 'y':
for q_id in failed:
savout = os.dup(1)
os.close(1)
os.open(os.devnull, os.O_RDWR)
try:
webbrowser.open('http://'+site+'.com/questions/'+str(q_id))
finally:
os.dup2(savout, 1)
kill_code('Finished Tagging')

if len(sys.argv) < 2:
site=input('Site: ').lower()
else:
site = sys.argv[1]

# Get the tag to remove from questions
if len(sys.argv) < 3:
tag=input('Tag: ').lower()
else:
tag = sys.argv[2]

# Get the number of questions to edit in batch. Max 30
if len(sys.argv) < 4:
try:
batch=min(30, int(input('Number to edit at once: ')))
except ValueError:
print ('Invalid input. Setting to 5')
batch=5
else:
try:
batch=min(30, int(sys.argv[3]))
except ValueError:
print ('Invalid input. Setting to 5')
batch=5

#print (batch)

# Get the number of
if len(sys.argv) < 5:
replacement_tags=input('Tag alternatives: ').lower().split()
else:
replacement_tags = sys.argv[4:]

# Check the replacement tags are valid (i.e. there is 1 or more)
if not replacement_tags:
print ('Invalid replacement tags. Exiting')
kill_code()

token = get_auth()
key = 'hDZI3p7wr3JAf1t)ccIIHA(('
items = get_from_search(site,batch,tag,key,token)
question_IDs = get_ids_from_items(items)
print ("Got IDs, fetching data")
all_data = get_tags_from_ids(question_IDs,key,token,batch)
formed_data=show_tags(all_data,tag,replacement_tags,site)
results=send_edits(formed_data,key,token,site,tag)
failed=results[0]
titles=results[1]
errors=results[2]
display_failed(failed,titles,errors,site)

• I'm tempted to DV on the moral grounds that this is not how retagging should be done, but ++ for the good question. – RubberDuck May 28 '15 at 22:11
• I'm just morally opposed to removing tags like this. IMO time should be taken to really clean up each post. Ask Ubuntu is none of my business though. You've run this past their meta, right? – RubberDuck May 29 '15 at 9:54
• @RubberDuck I spoke to a mod in chat, and he said it's fine, just keep it rate limited. I base it on the tags at first (like boot, uefi and live-cd suggest system and apt, software-center suggest software) but also use the question body if I can't tell. – Tim May 29 '15 at 9:56
• Cool. Just checking. – RubberDuck May 29 '15 at 10:02

### 1. Introduction

There's quite a bit of code here, so I'm just going to review one of your functions (get_width). You'll see that there's plenty here for one answer. Maybe some other reviewers will look at some of the rest of your code.

# uses os.popen to get the terminal size.
def get_width():
global width
try:
width = os.popen('stty size', 'r').read().split()[1]
supported=True
except IndexError as error:
width = 50
supported=False
width = int(width)
return supported


### 2. Review

1. The name of the function is ambiguous: get the width of what? Something like terminal_width would be better.

2. There's a comment but it would be better if this were a docstring: then you could read it using help(get_width) from the interactive interpreter.

3. The comment says:

# uses os.popen to get the terminal size.


but this is misleading. Someone who just read the comment might think that get_width returned the terminal width, but in fact the function stores the terminal width in a global variable and returns something else.

4. It's not clear what the "something else" is. The variable name is supported. But what is it that's supported or not?

5. It's a bad idea to use global variables to pass state around — it's all too easy to forget whether or not you've updated them. I presume that you're doing this because get_width has two pieces of information to return (the terminal width, and whether something or other is supported). But in Python, functions can return multiple values using a tuple, like this:

return supported, width

6. The code says:

except IndexError as error:


but error is not used, so write:

except IndexError:

7. There's no need to call int(width) in the IndexError case, as the code has already set width = 50.

8. If stty fails for some reason, for example:

stty: stdin isn't a terminal


then int(width) will raise ValueError. This needs to be handled.

9. stty size reports 0 0 if it can't determine the terminal size. This needs to be handled.

10. The mode argument to os.popen defaults to 'r', so there's no need to specify it.

11. The code opens a pipe using os.popen but does not close it. It happens to be the case that CPython automatically closes files when the reference count of the associated file object goes to zero, but this is not something you want to get into the habit of relying on because other Python implementations have different behaviour. Use the with statement to ensure that the pipe is closed:

with os.popen('stty size') as f:
try:
# etc.

12. os.popen runs a subprocess using a shell. This wastes a process (you don't need any shell features, you only want to run stty) and is a bad habit to get into because the shell has to parse the command you give it, and in more complicated situations that can be a security risk. It would be better to use subprocess.check_output to run stty directly, bypassing the shell.

13. Instead of calling the external process stty, why not interrogate the terminal directly? The idea is to use the ioctl (I/O device control) system call via the Python interface fnctl.ioctl, passing the TIOCGWINSZ ("terminal I/O control: get window size") command, which returns a pair of unsigned short (16-bit) values that can be decoded using struct.unpack.

rows, cols = unpack('hh', ioctl(sys.stdout, TIOCGWINSZ, '0000'))


This is exactly what the stty size command does, except that you are avoiding the extra process.

14. If you can't determine the terminal size via the ioctl call, it's a good idea to fall back to consulting the environment. Some shells set the LINES and COLUMNS environment variables.

15. To make the function more generally usable, why not return the rows as well as the columns? You have the value in your hand, so why not return it?

16. You use the supported flag to decide what kind of progress bar to draw. Presumably what you actually want to know is whether the standard output is connected to a terminal? But in that case you should just use the built-in os.isatty.

### 3. Revised code

from struct import unpack
from fcntl import ioctl
import os
import sys
from termios import TIOCGWINSZ

def terminal_size(fd=sys.stdout, default=(25, 80)):
"""Return the size of the terminal attached to fd, as a tuple (rows,
columns). If the size can't be determined, return default.

"""
def fallback(value, key, default):
if value > 0:
return value
try:
return int(os.environ.get(key, default))
except ValueError:
return default

size = unpack('hh', ioctl(fd, TIOCGWINSZ, '0000'))
return tuple(map(fallback, size, ('LINES', 'COLUMNS'), default))

• Thanks for this - I won't use all the advice but a lot of this is very helpful! – Tim May 29 '15 at 11:03

This is not about the functionality of the code, just it's presentation and readability.

You sometimes don't have spaces around the assignment (e.g. HTMLParser=HTMLParser instead of HTMLParser = HTMLParser).

That example shows some redundancy as well; HTMLParser is already itself, so there is no need to assign it.

And the way you have imported HTMLParser is not very well commented. I would go for something more like:

try:
import HTMLParser  # Python 2.x
except ImportError:
import html.parser as HTMLParser  # Python 3.x


Note this is an ImportError, not a NameError. You should test this with Python 3.

Most of your comments should really be docstrings, as in:

def spacer():
"""
Prints out a spacer like ---------- the width of the console.
If width is not supported, it will be 50 wide.
"""
get_width()
print('-' * width)


(Note that I added a linebreak to make it not exceed the line limit.)

Now I'm going to review one of your functions, loadin:

# Creates a loading bar like [========     ] the width of the console. Updates with get_width() each time it's called.
if get_width():
factor=(n/(width-2.0))
n=int(maths.floor((n)/factor))
i=int(maths.floor((i-1)/factor))
sys.stdout.write('['+'='*(n-i)+' '*(i)+']')
sys.stdout.flush()
sys.stdout.write('\r')
sys.stdout.flush()
else:
print ('=')


Now, I'm going to assume you are using Gareth Rees's terminal_size. The code needs to be adjusted to use it. I also accounted for the / 0 case and that int(math.floor(x)) = int(x) (if x is a float / int)

def loading_bar(n, i):
"""
Prints a loading bar like "[===    ]" if the terminal has width and
it is greater than two. Otherwise, it prints '='.
"""
if os.isatty(sys.stdout):
_, width = terminal_size()
n = float(n)  # To use float division
i = float(i)
try:
factor = n / (width - 2)
except ZeroDivisionError:
print('=')
return
n = int(n / factor)
i = int((i - 1) / factor)
sys.stdout.write('[' + '=' * (n - i) + ' ' * i + ']')
sys.stdout.flush()
sys.stdout.write('\r')
sys.stdout.flush()
else:
print('=')


Now your sys.stdout.write could really be merged into one write then flush.. Also, you could use str.format instead of concatenation.

        try:
sys.stdout.write('[{0}{1}]\r'.format('=' * (n - i), ' ' * i))
except AttributeError:  # Older Python 2 versions don't have string formatting
sys.stdout.write('[%s%s]\r' % ('=' * (n - i), ' ' * i))
sys.stdout.flush()


Now, the variable names aren't very specific. What are n and i? I think, as a loading bar, it should take a percentage the bar is full. Also, to avoid the string formatting and many concatenations, I would premake the bar in a variable then add the []. So, the final code using percentages would be:

def loading_bar(percentage_full):
"""
Prints a loading bar like "[===   ]" that is percentage_full% full.
If the terminal doesn't have a determinable width, prints "="
"""
if os.isatty(sys.stdout):
_, width = terminal_size()
if width <= 2:
print("=")
return
filled = math.round(width * percentage_full * 0.01)
# * 0.01 to turn percentage to decimal
whitespace = width - filled
bar = "=" * filled + " " * whitespace
sys.stdout.write("[" + bar + "]\r")
sys.stdout.flush()
else:
print("=")

• The reason for repeatedly calling get_width is that a terminal's width can change, for example, if the user resizes the window. (I guess this was not obvious from the code, so it probably deserves a comment.) – Gareth Rees May 29 '15 at 10:50
• @GarethRees I see your answer heavily reviewed that function. I have removed it from mine because df my misunderstanding – user61114 May 29 '15 at 10:54
• @Reticality Thanks for this answer. I assigned HTMLParser to itself for python 2 and 3 support - fixed that now. I will read through the answer when I'm back online this evening (UTC). – Tim May 29 '15 at 11:54
• What is bar_length? Also it can't be print because otherwise it doesn't overwrite the previous line... – Tim May 30 '15 at 11:43
• @Tim Sorry, I renamed it to filled but forgot to change it. I have changed it – user61114 May 31 '15 at 18:12

Okay, there are a couple of WTFs with the code. I'll leave out the already commented-on stuff.

• What/why is there global HTML and HTML=max in there? That doesn't do anything except give it a new name. In which case the global is still not needed. IMO just don't do that. Same goes for all the renamed imports, i.e. update a.k.a. sleep, maths a.k.a. math. The point is that the reader with higher probability knows what math, sleep and max are. It is way easier to just keep the names instead of having to look up trivial name changes.

• What is True*8**2-4? I now know it's 60. How about writing 60 instead and not keep the reader guessing what a boolean times an integer is.

• kill_code is overkill. Either of those functions will exit the process. So just replace kill_code with sys.exit. Stack Overflow agrees with me.

• The return value for check_for_error is never used, you can just drop that in favour of pass. Then again, it's cleaner to check for the existence of the key instead of catching the KeyError instead. In general it's also not good practice to kill the process if a minor condition fails, e.g. raising an exception (which in turn could be handled, or otherwise quit the process) is more future-proof. I'd expect at least something like the following instead (modulo the exception thing):

def check_for_error(response_JSON):
if 'error_message' in response_JSON:
sys.exit(response_JSON['error_message'])

• The 'authtoken.txt' opened in get_auth is never closed. Use with as well, otherwise you're leaking file descriptors and that's never good.

• get_auth is called recursively, but the recursive call has no return, so if this fails on the first try it will return None regardless.

Minor things:

• input = input doesn't do anything.
• String formatting is easier with interpolation or .format.
• THere's both unicode and regular string literals used. If possible use just of both forms.
• get_ids_from_items can be simplified:

def get_ids_from_items(items):
return [item['question_id'] for item in items]


That also avoids creating all those single element lists. Since that appears a number of times then other option there is to use .append instead (for the same reason).

• In change_tag setting tag_id to zero doesn't do anything.
• quota_remaining in get_tags_from_ids is unused.
• The iteration in print_tags can be made a little bit nicer with enumerate instead.
• Move everything on the top level into a if __name__ == '__main__': block, or even a main function. It also wouldn't hurt to use argparse for command-line arguments.
• The sequence response.json(), backoff(), check_for_error is used three times - I'd create a new function for it.

In general it would be good to have more code reuse and reduce clutter with all the mixing of UI messages and actual work. More functions for formatting stuff would also reduce the number of lines dedicated to string manipulation.

Also some comments in the important places would be nice, i.e. why the sudden low-level file descriptor manipulation with os.dup` and friends is necessary.